hcq/oceanbase
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity
Files
generic
oceanbase/unittest/storage/blocksstable/test_row_reader.cpp
gm 4a92b6d7df reformat source code 
according to code styles, 'AccessModifierOffset' should be -2.
2021-06-17 10:40:36 +08:00

2161 lines
73 KiB
C++
Raw Permalink Blame History

/**
* Copyright (c) 2021 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.
*/
#include <gtest/gtest.h>
#include "storage/blocksstable/ob_row_writer.h"
#include "storage/blocksstable/ob_row_reader.h"
#include "storage/ob_i_store.h"
#include "storage/blocksstable/ob_block_sstable_struct.h"
#include "lib/number/ob_number_v2.h"
#include "ob_row_generate.h"
#include "share/object/ob_obj_cast.h"
#include "storage/blocksstable/ob_column_map.h"
#define protected public
#define private public
#ifndef INT24_MIN
#define INT24_MIN (-8388607 - 1)
#endif
#ifndef INT24_MAX
#define INT24_MAX (8388607)
#endif
#ifndef UINT24_MAX
#define UINT24_MAX (16777215U)
#endif
namespace oceanbase {
using namespace common;
using namespace blocksstable;
using namespace storage;
using namespace common::number;
using namespace share::schema;
namespace unittest {
class TestRowReader : public ::testing::Test {
public:
static const int64_t rowkey_column_count = 1;
// Every ObObjType from ObTinyIntType to ObHexStringType inclusive.
// Skip ObNullType and ObExtendType because for external usage, a column type
// can't be NULL or NOP.
ObObjType not_test_type[8] = {ObNullType,
ObExtendType,
ObUnknownType,
ObEnumInnerType,
ObSetInnerType,
ObNumberFloatType,
ObURowIDType,
ObLobType};
static const int64_t column_num = ObMaxType - sizeof(not_test_type) / sizeof(ObObjType);
public:
virtual void SetUp();
virtual void TearDown();
char* get_serialize_buf()
{
return serialize_buf_;
}
private:
int build_column_map(const int64_t* columns, const int64_t count);
int init_column_map(const ObStoreRow& writer_row, ObArray<ObColDesc>& columns);
char* serialize_buf_;
protected:
ObRowGenerate row_generate_;
ObColumnMap column_map_;
ObTableSchema table_schema_;
ObArenaAllocator allocator_;
};
int TestRowReader::init_column_map(const ObStoreRow& writer_row, ObArray<ObColDesc>& columns)
{
int ret = OB_SUCCESS;
for (int64_t i = 0; i < column_num; ++i) {
ObObjType obj_type = writer_row.row_val_.cells_[i].get_type();
ObObjMeta obj_meta;
obj_meta.set_type(obj_type);
if (ObVarcharType == obj_type || ObCharType == obj_type || ObNCharType == obj_type || ObNVarchar2Type == obj_type ||
ObTinyTextType == obj_type || ObTextType == obj_type || ObMediumTextType == obj_type ||
ObLongTextType == obj_type) {
obj_meta.set_collation_type(CS_TYPE_UTF8MB4_GENERAL_CI);
obj_meta.set_collation_level(CS_LEVEL_IMPLICIT);
} else if (ObRawType == obj_type) {
obj_meta.set_collation_type(CS_TYPE_BINARY);
}
columns[i].col_type_ = obj_meta;
}
if (OB_FAIL(column_map_.init(allocator_,
row_generate_.get_schema().get_schema_version(),
row_generate_.get_schema().get_rowkey_column_num(),
column_num,
columns))) {
STORAGE_LOG(WARN, "failed to init column map");
}
return ret;
}
int TestRowReader::build_column_map(const int64_t* columns, const int64_t count)
{
int ret = OB_SUCCESS;
column_map_.reset();
ObArray<ObColDesc> out_cols;
ObArray<int32_t> projector;
for (int64_t i = 0; i < count; ++i) {
int64_t request_id = columns[i];
const ObColumnSchemaV2* column = table_schema_.get_column_schema(request_id);
int64_t store_index = table_schema_.get_column_idx(request_id, true);
if (NULL == column)
return OB_ERROR;
ObColDesc col_desc;
col_desc.col_id_ = request_id;
col_desc.col_type_ = column->get_meta_type();
out_cols.push_back(col_desc);
projector.push_back(static_cast<int32_t>(store_index));
}
ret = column_map_.init(allocator_,
table_schema_.get_schema_version(),
table_schema_.get_rowkey_column_num(),
column_num,
out_cols,
nullptr,
&projector);
if (!column_map_.is_valid())
return OB_NOT_INIT;
return ret;
}
void TestRowReader::SetUp()
{
const int64_t serialize_buf_size_ = 10 * 1024 * 1024; // 2M
serialize_buf_ = reinterpret_cast<char*>(ob_malloc(serialize_buf_size_, ObModIds::TEST));
ASSERT_TRUE(NULL != serialize_buf_);
const int64_t table_id = 3001;
oceanbase::common::ObClusterVersion::get_instance().refresh_cluster_version("1.4.70");
ObColumnSchemaV2 column;
// init table schema
table_schema_.reset();
ASSERT_EQ(OB_SUCCESS, table_schema_.set_table_name("test_row_writer"));
table_schema_.set_tenant_id(1);
table_schema_.set_tablegroup_id(1);
table_schema_.set_database_id(1);
table_schema_.set_table_id(table_id);
table_schema_.set_rowkey_column_num(rowkey_column_count);
table_schema_.set_max_used_column_id(column_num);
// init column
char name[OB_MAX_FILE_NAME_LENGTH];
memset(name, 0, sizeof(name));
int not_test_index = 0;
for (int64_t i = 0; i < ObMaxType; ++i) {
ObObjType obj_type = static_cast<ObObjType>(i);
if (not_test_type[not_test_index] == obj_type) {
not_test_index++;
continue;
}
column.reset();
column.set_table_id(table_id);
column.set_column_id(i + OB_APP_MIN_COLUMN_ID);
sprintf(name, "test%020ld", i);
ASSERT_EQ(OB_SUCCESS, column.set_column_name(name));
column.set_data_type(obj_type);
column.set_collation_type(CS_TYPE_UTF8MB4_GENERAL_CI);
if (obj_type == common::ObIntType) {
column.set_rowkey_position(1);
} else {
column.set_rowkey_position(0);
}
ASSERT_EQ(OB_SUCCESS, table_schema_.add_column(column));
}
// init ObRowGenerate
ASSERT_EQ(OB_SUCCESS, row_generate_.init(table_schema_));
}
void TestRowReader::TearDown()
{
ob_free(serialize_buf_);
}
void append_col(ObStoreRow& row)
{
int index = row.row_val_.count_;
ObScale scale;
scale = ObLobScale::STORE_IN_ROW;
row.row_val_.cells_[index].meta_.set_collation_type(CS_TYPE_UTF8MB4_GENERAL_CI);
row.row_val_.cells_[index].meta_.set_collation_level(CS_LEVEL_IMPLICIT);
row.row_val_.cells_[index].set_scale(scale);
row.row_val_.cells_[index++].set_string(ObTinyTextType, "xxxxxx");
row.row_val_.cells_[index].meta_.set_collation_type(CS_TYPE_UTF8MB4_GENERAL_CI);
row.row_val_.cells_[index].meta_.set_collation_level(CS_LEVEL_IMPLICIT);
row.row_val_.cells_[index].set_scale(scale);
row.row_val_.cells_[index++].set_string(ObTextType, "xxxxxx");
row.row_val_.cells_[index].meta_.set_collation_type(CS_TYPE_UTF8MB4_GENERAL_CI);
row.row_val_.cells_[index].meta_.set_collation_level(CS_LEVEL_IMPLICIT);
row.row_val_.cells_[index].set_scale(scale);
row.row_val_.cells_[index++].set_string(ObMediumTextType, "xxxxxx");
row.row_val_.cells_[index].meta_.set_collation_type(CS_TYPE_UTF8MB4_GENERAL_CI);
row.row_val_.cells_[index].meta_.set_collation_level(CS_LEVEL_IMPLICIT);
row.row_val_.cells_[index].set_scale(scale);
row.row_val_.cells_[index++].set_string(ObLongTextType, "xxxxxx");
row.row_val_.cells_[index++].set_bit(999);
row.row_val_.cells_[index++].set_enum(88);
row.row_val_.cells_[index++].set_set(1849);
// row.row_val_.cells_[index++].set_enum_inner(ObString("jakgja"));
// row.row_val_.cells_[index++].set_set_inner(ObString("jfao"));
uint16_t sec = 12;
row.row_val_.cells_[index++].set_otimestamp_value(ObTimestampTZType, 888, sec);
row.row_val_.cells_[index++].set_otimestamp_value(ObTimestampLTZType, 888, sec);
row.row_val_.cells_[index++].set_otimestamp_value(ObTimestampNanoType, 888, sec);
row.row_val_.cells_[index++].set_raw(ObString("agjaljg"));
row.row_val_.cells_[index++].set_interval_ym(ObIntervalYMValue(11));
row.row_val_.cells_[index++].set_interval_ds(ObIntervalDSValue(11, 14));
uint32_t digits = 8;
row.row_val_.cells_[index++].set_number_float(ObNumber(3, &digits));
row.row_val_.cells_[index].meta_.set_collation_type(CS_TYPE_UTF8MB4_GENERAL_CI);
row.row_val_.cells_[index].meta_.set_collation_level(CS_LEVEL_IMPLICIT);
row.row_val_.cells_[index++].set_nvarchar2(ObString("ahgiqh"));
row.row_val_.cells_[index].meta_.set_collation_type(CS_TYPE_UTF8MB4_GENERAL_CI);
row.row_val_.cells_[index].meta_.set_collation_level(CS_LEVEL_IMPLICIT);
row.row_val_.cells_[index++].set_nchar(ObString("ahkqh"));
for (int i = 0; i < index; ++i) {
row.column_ids_[i] = 100 + i;
}
row.row_val_.count_ = index;
}
TEST_F(TestRowReader, test_number_float)
{
ObObj write_obj;
uint32_t digits = 8;
write_obj.set_number_float(ObNumber(3, &digits));
char* buf = get_serialize_buf();
int64_t buf_size = 2 * 1024 * 1024;
int64_t pos = 0;
int ret = OB_SUCCESS;
ret = write_obj.get_unumber().encode(buf, buf_size, pos);
ASSERT_EQ(OB_SUCCESS, ret);
COMMON_LOG(INFO, "write number float", K(ret), K(write_obj), K(write_obj.get_unumber()));
ObObj read_obj;
number::ObNumber value;
int64_t read_pos = 0;
if (OB_FAIL(value.decode(buf, pos, read_pos))) {
COMMON_LOG(WARN, "decode number failed", K(ret));
} else {
read_obj.set_number_float(value);
COMMON_LOG(INFO, "read number float", K(ret), K(read_obj), K(value));
}
ASSERT_TRUE(read_obj == write_obj);
/*
// try in rowWriter
int num = 50;
ObRowWriter row_writer;
oceanbase::common::ObObj objs[num];
uint16_t column_ids[num];
ObStoreRow writer_row;
writer_row.row_val_.cells_ = objs;
writer_row.row_val_.count_ = column_num;
writer_row.column_ids_ = column_ids;
writer_row.flag_ = ObActionFlag::OP_ROW_DOES_NOT_EXIST;
writer_row.dml_ = T_DML_INSERT;
digits = 0;
common::ObObjMeta column_type_array[column_num];
ObObjMeta obj_meta;
obj_meta.set_type(ObNumberFloatType);
for(int i = 0; i < column_num; ++i) {
digits = i;
writer_row.row_val_.cells_[i].set_number_float(ObNumber(i, &digits));
column_type_array[i] = obj_meta;
}
int64_t rowkey_start_pos = 0;
int64_t rowkey_length = 0;
pos = 0;
ret = row_writer.write(1, writer_row, buf, 2 * 1024 * 1024,
pos, rowkey_start_pos, rowkey_length);
ASSERT_EQ(OB_SUCCESS, ret);
STORAGE_LOG(INFO, "write_row", K(writer_row));
ObFlatRowReader row_reader;
ObStoreRow reader_row;
oceanbase::common::ObObj read_objs[num];
uint16_t read_column_ids[num];
reader_row.row_val_.cells_ = read_objs;
reader_row.row_val_.count_ = column_num;
reader_row.capacity_ = num;
reader_row.column_ids_ = read_column_ids;
ret = row_reader.read_full_row(buf, pos, read_pos, column_type_array, allocator_, reader_row);
ASSERT_EQ(OB_SUCCESS, ret);
STORAGE_LOG(INFO, "reader_row", K(reader_row));
for (int i = 0; i < column_num; ++i) {
ASSERT_EQ(writer_row.row_val_.cells_[i], reader_row.row_val_.cells_[i]);
}
*/
}
TEST_F(TestRowReader, test_cell_reader)
{
int ret = OB_SUCCESS;
int num = 50;
ObRowWriter row_writer;
oceanbase::common::ObObj objs[num];
uint16_t column_ids[num];
ObStoreRow writer_row;
writer_row.row_val_.cells_ = objs;
writer_row.row_val_.count_ = column_num;
writer_row.column_ids_ = column_ids;
ObSparseRowReader row_reader;
ObStoreRow reader_row;
oceanbase::common::ObObj read_objs[num];
uint16_t read_column_ids[num];
reader_row.row_val_.cells_ = read_objs;
reader_row.row_val_.count_ = num;
reader_row.capacity_ = num;
reader_row.column_ids_ = read_column_ids;
// init ObRowGenerate
int64_t pos = 0;
int64_t rowkey_start_pos = 0;
int64_t rowkey_length = 0;
int64_t read_pos = 0;
for (int i = 0; i < 10; ++i) {
writer_row.row_val_.count_ = column_num;
ASSERT_EQ(OB_SUCCESS, row_generate_.get_next_row(writer_row));
writer_row.is_sparse_row_ = true;
STORAGE_LOG(INFO, "write_row", K(writer_row));
pos = 0;
rowkey_start_pos = 0;
rowkey_length = 0;
char* buf = get_serialize_buf();
ret = row_writer.write(1, writer_row, buf, 2 * 1024 * 1024, pos, rowkey_start_pos, rowkey_length);
ASSERT_EQ(OB_SUCCESS, ret);
read_pos = 0;
ret = row_reader.read_full_row(buf, pos, read_pos, nullptr, allocator_, reader_row);
ASSERT_EQ(OB_SUCCESS, ret);
STORAGE_LOG(INFO, "read_full_row", K(reader_row));
ASSERT_EQ(reader_row.row_val_.count_, writer_row.row_val_.count_);
ASSERT_EQ(reader_row.is_sparse_row_, writer_row.is_sparse_row_);
for (int j = 0; j < reader_row.row_val_.count_; ++j) {
STORAGE_LOG(INFO, "obj", K(reader_row.row_val_.cells_[j]), K(writer_row.row_val_.cells_[j]));
if (ObNumberFloatType != reader_row.row_val_.cells_[j].get_type()) {
ASSERT_TRUE(reader_row.row_val_.cells_[j] == writer_row.row_val_.cells_[j]);
ASSERT_TRUE(reader_row.column_ids_[j] == writer_row.column_ids_[j]);
}
}
}
}
TEST_F(TestRowReader, init_fail)
{
int ret = OB_SUCCESS;
ObFlatRowReader row_reader;
ObStoreRow reader_row;
oceanbase::common::ObObj objs[column_num];
reader_row.row_val_.cells_ = objs;
reader_row.row_val_.count_ = column_num;
reader_row.capacity_ = column_num;
int64_t pos = 0;
int64_t rowkey_start_pos = 0;
int64_t rowkey_length = 0;
int64_t read_pos = 0;
char* buf = get_serialize_buf();
pos = 0;
rowkey_start_pos = 0;
rowkey_length = 0;
read_pos = 0;
// invalid column map
ret = row_reader.read_row(buf, pos, read_pos, column_map_, allocator_, reader_row);
ASSERT_EQ(OB_INVALID_ARGUMENT, ret);
// invalid buf
column_map_.reset();
ObArray<ObColDesc> columns;
ASSERT_EQ(OB_SUCCESS, row_generate_.get_schema().get_column_ids(columns));
for (int64_t i = 0; i < column_num; ++i) {
ObObjMeta obj_meta;
ObObjType obj_type = static_cast<ObObjType>(i);
obj_meta.set_type(obj_type);
if (ObVarcharType == obj_type || ObCharType == obj_type) {
obj_meta.set_collation_type(CS_TYPE_UTF8MB4_GENERAL_CI);
obj_meta.set_collation_level(CS_LEVEL_IMPLICIT);
}
columns[i].col_type_ = obj_meta;
}
share::schema::ColumnMap col_id_map(allocator_);
ret = col_id_map.init(columns);
ASSERT_EQ(OB_SUCCESS, ret);
ASSERT_EQ(OB_SUCCESS,
column_map_.init(allocator_,
row_generate_.get_schema().get_schema_version(),
row_generate_.get_schema().get_rowkey_column_num(),
column_num,
columns));
ret = row_reader.read_row(NULL, pos, read_pos, column_map_, allocator_, reader_row);
ASSERT_EQ(OB_INVALID_ARGUMENT, ret);
// invalid pos
pos = 10;
read_pos = 100;
ret = row_reader.read_row(buf, pos, read_pos, column_map_, allocator_, reader_row);
ASSERT_EQ(OB_INVALID_ARGUMENT, ret);
pos = 0;
ret = row_reader.read_row(buf, pos, read_pos, column_map_, allocator_, reader_row);
ASSERT_EQ(OB_INVALID_ARGUMENT, ret);
pos = 100;
read_pos = -1;
ret = row_reader.read_row(buf, pos, read_pos, column_map_, allocator_, reader_row);
ASSERT_EQ(OB_INVALID_ARGUMENT, ret);
(void)rowkey_start_pos;
(void)rowkey_length;
}
TEST_F(TestRowReader, read_flat_row_from_flat_storage)
{
int ret = OB_SUCCESS;
ObRowWriter row_writer;
oceanbase::common::ObObj objs[column_num];
ObStoreRow writer_row;
writer_row.row_val_.cells_ = objs;
writer_row.row_val_.count_ = column_num;
ObFlatRowReader row_reader;
ObStoreRow reader_row;
oceanbase::common::ObObj reader_objs[column_num];
reader_row.row_val_.cells_ = reader_objs;
reader_row.row_val_.count_ = column_num;
reader_row.capacity_ = column_num;
int64_t pos = 0;
int64_t rowkey_start_pos = 0;
int64_t rowkey_length = 0;
int64_t read_pos = 0;
char* buf = get_serialize_buf();
// write success
pos = 0;
rowkey_start_pos = 0;
rowkey_length = 0;
ASSERT_EQ(OB_SUCCESS, row_generate_.get_next_row(writer_row));
ret = row_writer.write_flat_row(
rowkey_column_count, writer_row, buf, 2 * 1024 * 1024, pos, rowkey_start_pos, rowkey_length);
ASSERT_EQ(OB_SUCCESS, ret);
// init ObColumnMap
column_map_.reset();
ObArray<ObColDesc> columns;
ASSERT_EQ(OB_SUCCESS, row_generate_.get_schema().get_column_ids(columns));
ret = init_column_map(writer_row, columns);
ASSERT_EQ(OB_SUCCESS, ret);
// read success
read_pos = 0;
ret = row_reader.read_row(buf, pos, read_pos, column_map_, allocator_, reader_row);
ASSERT_EQ(OB_SUCCESS, ret);
STORAGE_LOG(INFO, "read row", K(reader_row));
bool exist = false;
ASSERT_EQ(OB_SUCCESS, row_generate_.check_one_row(reader_row, exist));
ASSERT_TRUE(column_num == reader_row.row_val_.count_);
// every obj should equal
for (int64_t i = 0; i < column_num; ++i) {
STORAGE_LOG(INFO, "compare", K(i), K(writer_row.row_val_.cells_[i]), K(reader_row.row_val_.cells_[i]));
ASSERT_EQ(writer_row.row_val_.cells_[i], reader_row.row_val_.cells_[i]);
}
// new row write success
ObNewRow new_reader_row;
new_reader_row.count_ = column_num;
new_reader_row.cells_ = reader_objs;
pos = 0;
ASSERT_EQ(OB_SUCCESS, row_generate_.get_next_row(writer_row));
ret = row_writer.write(writer_row.row_val_, buf, 2 * 1024 * 1024, FLAT_ROW_STORE, pos);
ASSERT_EQ(OB_SUCCESS, ret);
// init column type array
ObObjMeta column_types[common::OB_ROW_MAX_COLUMNS_COUNT];
for (int64_t i = 0; i < column_num; ++i) {
column_types[i] = writer_row.row_val_.cells_[i].get_meta();
}
// new row read success
read_pos = 0;
ret =
row_reader.read_compact_rowkey(column_types, column_map_.get_request_count(), buf, pos, read_pos, new_reader_row);
ASSERT_EQ(OB_SUCCESS, ret);
// every obj should equal
for (int64_t i = 0; i < column_num; ++i) {
ASSERT_TRUE(writer_row.row_val_.cells_[i] == new_reader_row.cells_[i])
<< "\n i: " << i << "\n writer: " << to_cstring(writer_row.row_val_.cells_[i])
<< "\n reader: " << to_cstring(reader_row.row_val_.cells_[i]);
}
}
TEST_F(TestRowReader, read_flat_row_from_flat_storage2)
{
int ret = OB_SUCCESS;
ObRowWriter row_writer;
oceanbase::common::ObObj objs[column_num];
ObStoreRow writer_row;
writer_row.row_val_.cells_ = objs;
writer_row.row_val_.count_ = column_num;
ObFlatRowReader row_reader;
ObStoreRow reader_row;
oceanbase::common::ObObj reader_objs[column_num];
reader_row.row_val_.cells_ = reader_objs;
reader_row.row_val_.count_ = column_num;
reader_row.capacity_ = column_num;
int64_t pos = 0;
int64_t rowkey_start_pos = 0;
int64_t rowkey_length = 0;
int64_t read_pos = 0;
char* buf = get_serialize_buf();
// write success
pos = 0;
rowkey_start_pos = 0;
rowkey_length = 0;
ASSERT_EQ(OB_SUCCESS, row_generate_.get_next_row(writer_row));
// write 100 rows
ret = row_writer.write_flat_row(
rowkey_column_count, writer_row, buf, 2 * 1024 * 1024, pos, rowkey_start_pos, rowkey_length);
ASSERT_EQ(OB_SUCCESS, ret);
// init ObColumnMap
column_map_.reset();
ObArray<ObColDesc> columns;
ASSERT_EQ(OB_SUCCESS, row_generate_.get_schema().get_column_ids(columns));
ret = init_column_map(writer_row, columns);
ASSERT_EQ(OB_SUCCESS, ret);
// read success
read_pos = 0;
ret = row_reader.read_row(buf, pos, read_pos, column_map_, allocator_, reader_row, SPARSE_ROW_STORE);
ASSERT_EQ(OB_SUCCESS, ret);
ASSERT_TRUE(column_num == reader_row.row_val_.count_);
// every obj should equal
int index = 0;
for (int64_t i = 0; i < column_num; ++i) {
if (!writer_row.row_val_.cells_[i].is_nop_value()) {
ASSERT_TRUE(writer_row.row_val_.cells_[i] == reader_row.row_val_.cells_[index]);
ASSERT_TRUE(columns[i].col_id_ == reader_row.column_ids_[index]);
index++;
}
}
}
TEST_F(TestRowReader, read_overflow)
{
int ret = OB_SUCCESS;
ObRowWriter row_writer;
oceanbase::common::ObObj objs[column_num];
ObStoreRow writer_row;
writer_row.row_val_.cells_ = objs;
writer_row.row_val_.count_ = column_num;
ObFlatRowReader row_reader;
ObStoreRow reader_row;
oceanbase::common::ObObj reader_objs[column_num];
reader_row.row_val_.cells_ = reader_objs;
reader_row.row_val_.count_ = column_num;
reader_row.capacity_ = column_num;
int64_t pos = 0;
int64_t rowkey_start_pos = 0;
int64_t rowkey_length = 0;
int64_t read_pos = 0;
char* buf = get_serialize_buf();
// write success
ASSERT_EQ(OB_SUCCESS, row_generate_.get_next_row(writer_row));
ret = row_writer.write_flat_row(
rowkey_column_count, writer_row, buf, 2 * 1024 * 1024, pos, rowkey_start_pos, rowkey_length);
ASSERT_EQ(OB_SUCCESS, ret);
// init ObColumnMap
column_map_.reset();
ObArray<ObColDesc> columns;
ASSERT_EQ(OB_SUCCESS, row_generate_.get_schema().get_column_ids(columns));
ret = init_column_map(writer_row, columns);
ASSERT_EQ(OB_SUCCESS, ret);
// read_header_overflow
read_pos = 0;
ret = row_reader.read_row(buf, 3, read_pos, column_map_, allocator_, reader_row);
ASSERT_EQ(OB_SIZE_OVERFLOW, ret);
// read_column_index_overflow
read_pos = 0;
ret = row_reader.read_row(buf, sizeof(ObRowHeader), read_pos, column_map_, allocator_, reader_row);
ASSERT_EQ(OB_SIZE_OVERFLOW, ret);
// step == 1, read_columns overflow
read_pos = 0;
ret = row_reader.read_row(buf, 200, read_pos, column_map_, allocator_, reader_row);
ASSERT_EQ(OB_BUF_NOT_ENOUGH, ret);
// read_sequence_columns overflow
ObNewRow new_writer_row;
new_writer_row.count_ = column_num;
new_writer_row.cells_ = objs;
ObNewRow new_reader_row;
new_reader_row.count_ = 0;
new_reader_row.cells_ = reader_objs;
pos = 0;
ASSERT_EQ(OB_SUCCESS, row_generate_.get_next_row(writer_row));
ret = row_writer.write(new_writer_row, buf, 2 * 1024 * 1024, FLAT_ROW_STORE, pos);
ASSERT_EQ(OB_SUCCESS, ret);
// init column type array
ObObjMeta column_types[common::OB_ROW_MAX_COLUMNS_COUNT];
for (int64_t i = 0; i < column_num; ++i) {
ObObjType obj_type = new_writer_row.cells_[i].get_type();
ObObjMeta& obj_meta = column_types[i];
obj_meta.set_type(obj_type);
if (ObVarcharType == obj_type || ObCharType == obj_type) {
obj_meta.set_collation_type(CS_TYPE_UTF8MB4_GENERAL_CI);
obj_meta.set_collation_level(CS_LEVEL_IMPLICIT);
}
}
// new row read success
read_pos = 0;
ret =
row_reader.read_compact_rowkey(column_types, column_map_.get_request_count(), buf, 200, read_pos, new_reader_row);
ASSERT_EQ(OB_BUF_NOT_ENOUGH, ret);
}
TEST_F(TestRowReader, read_sparse_row_from_flat_storage)
{
int ret = OB_SUCCESS;
ObRowWriter row_writer;
oceanbase::common::ObObj objs[column_num];
ObStoreRow writer_row;
writer_row.row_val_.cells_ = objs;
writer_row.row_val_.count_ = column_num;
ObFlatRowReader row_reader;
ObStoreRow reader_row;
oceanbase::common::ObObj reader_objs[column_num];
reader_row.row_val_.cells_ = reader_objs;
reader_row.row_val_.count_ = column_num;
reader_row.capacity_ = column_num;
int64_t pos = 0;
int64_t rowkey_start_pos = 0;
int64_t rowkey_length = 0;
int64_t read_pos = 0;
char* buf = get_serialize_buf();
// write success
pos = 0;
rowkey_start_pos = 0;
rowkey_length = 0;
ASSERT_EQ(OB_SUCCESS, row_generate_.get_next_row(writer_row));
// set nop value
int nop_cell_cnt = 5;
int nop_indexs[] = {5, 9, 12, 18, 21};
for (int i = 0; i < nop_cell_cnt; ++i) {
writer_row.row_val_.cells_[nop_indexs[i]].set_nop_value();
}
ret = row_writer.write_flat_row(
rowkey_column_count, writer_row, buf, 2 * 1024 * 1024, pos, rowkey_start_pos, rowkey_length);
ASSERT_EQ(OB_SUCCESS, ret);
// init ObColumnMap
column_map_.reset();
ObArray<ObColDesc> columns;
ASSERT_EQ(OB_SUCCESS, row_generate_.get_schema().get_column_ids(columns));
ret = init_column_map(writer_row, columns);
ASSERT_EQ(OB_SUCCESS, ret);
// read success
read_pos = 0;
ret = row_reader.read_row(buf, pos, read_pos, column_map_, allocator_, reader_row, SPARSE_ROW_STORE);
ASSERT_EQ(OB_SUCCESS, ret);
ASSERT_TRUE(column_num - nop_cell_cnt == reader_row.row_val_.count_);
// every obj should equal
int index = 0;
for (int64_t i = 0; i < column_num; ++i) {
if (!writer_row.row_val_.cells_[i].is_nop_value()) {
ASSERT_TRUE(writer_row.row_val_.cells_[i] == reader_row.row_val_.cells_[index]);
ASSERT_TRUE(columns[i].col_id_ == reader_row.column_ids_[index]);
index++;
}
}
}
TEST_F(TestRowReader, flat_row_test)
{
int ret = OB_SUCCESS;
ObRowWriter row_writer;
oceanbase::common::ObObj objs[column_num];
ObStoreRow writer_row;
writer_row.row_val_.cells_ = objs;
writer_row.row_val_.count_ = column_num;
ObFlatRowReader row_reader;
ObStoreRow reader_row;
oceanbase::common::ObObj reader_objs[column_num];
reader_row.row_val_.cells_ = reader_objs;
reader_row.row_val_.count_ = column_num;
reader_row.capacity_ = column_num;
int64_t pos = 0;
int64_t rowkey_start_pos = 0;
int64_t rowkey_length = 0;
int64_t read_pos = 0;
char* buf = get_serialize_buf();
// write success
pos = 0;
rowkey_start_pos = 0;
rowkey_length = 0;
ASSERT_EQ(OB_SUCCESS, row_generate_.get_next_row(writer_row));
// set nop value
int nop_cell_cnt = 5;
int nop_indexs[] = {5, 9, 12, 18, 21};
for (int i = 0; i < nop_cell_cnt; ++i) {
writer_row.row_val_.cells_[nop_indexs[i]].set_nop_value();
}
// write 100 rows
int row_cnt = 100;
for (int i = 0; i < row_cnt; ++i) {
ret = row_writer.write_flat_row(
rowkey_column_count, writer_row, buf, 2 * 1024 * 1024, pos, rowkey_start_pos, rowkey_length);
ASSERT_EQ(OB_SUCCESS, ret);
}
// init ObColumnMap
column_map_.reset();
ObArray<ObColDesc> columns;
ASSERT_EQ(OB_SUCCESS, row_generate_.get_schema().get_column_ids(columns));
ret = init_column_map(writer_row, columns);
ASSERT_EQ(OB_SUCCESS, ret);
// read success
read_pos = 0;
for (int j = 0; j < row_cnt; ++j) {
ret = row_reader.read_row(buf, pos, read_pos, column_map_, allocator_, reader_row, SPARSE_ROW_STORE);
ASSERT_EQ(OB_SUCCESS, ret);
ASSERT_TRUE(column_num - nop_cell_cnt == reader_row.row_val_.count_);
// every obj should equal
int index = 0;
for (int64_t i = 0; i < column_num; ++i) {
if (!writer_row.row_val_.cells_[i].is_nop_value()) {
ASSERT_TRUE(writer_row.row_val_.cells_[i] == reader_row.row_val_.cells_[index]);
ASSERT_TRUE(columns[i].col_id_ == reader_row.column_ids_[index]);
index++;
}
}
}
}
TEST_F(TestRowReader, flat_row_test_about_trans_id_flag_change)
{
int ret = OB_SUCCESS;
ObRowWriter row_writer;
oceanbase::common::ObObj objs[column_num];
ObStoreRow writer_row;
writer_row.row_val_.cells_ = objs;
writer_row.row_val_.count_ = column_num;
ObFlatRowReader row_reader;
ObStoreRow reader_row;
oceanbase::common::ObObj reader_objs[column_num];
reader_row.row_val_.cells_ = reader_objs;
reader_row.row_val_.count_ = column_num;
reader_row.capacity_ = column_num;
int64_t pos = 0;
int64_t rowkey_start_pos = 0;
int64_t rowkey_length = 0;
int64_t read_pos = 0;
char* buf = get_serialize_buf();
// write success
pos = 0;
rowkey_start_pos = 0;
rowkey_length = 0;
ASSERT_EQ(OB_SUCCESS, row_generate_.get_next_row(writer_row));
int idx = 0;
writer_row.row_val_.cells_[idx++].set_int32(1);
writer_row.row_val_.cells_[idx++].set_int(-INT64_MAX);
writer_row.row_val_.cells_[idx++].set_int(-36202);
writer_row.row_val_.cells_[idx++].set_int32(1);
transaction::ObTransID trans_id(ObAddr(ObAddr::IPV4, "11.158.41.173", 16046));
writer_row.row_type_flag_.flag_ = 10;
writer_row.trans_id_ptr_ = &trans_id;
writer_row.row_val_.count_ = 4;
// write 100 rows
ret = row_writer.write_flat_row(
rowkey_column_count, writer_row, buf, 2 * 1024 * 1024, pos, rowkey_start_pos, rowkey_length);
ASSERT_EQ(OB_SUCCESS, ret);
// init ObColumnMap
// read success
read_pos = 0;
transaction::ObTransID read_trans_id;
ret = row_reader.setup_row(buf, pos, read_pos, 4, &read_trans_id);
ASSERT_EQ(OB_SUCCESS, ret);
STORAGE_LOG(INFO, "read row", K(read_trans_id));
}
TEST_F(TestRowReader, read_flat_row_from_sparse_storage)
{
int ret = OB_SUCCESS;
ObRowWriter row_writer;
oceanbase::common::ObObj objs[column_num];
ObStoreRow writer_row;
writer_row.row_val_.cells_ = objs;
writer_row.row_val_.count_ = column_num;
ObSparseRowReader sparse_row_reader;
ObFlatRowReader flat_row_reader;
ObStoreRow sparse_row;
oceanbase::common::ObObj sparse_row_objs[column_num];
uint16_t sparse_col_id_array[column_num];
sparse_row.row_val_.cells_ = sparse_row_objs;
sparse_row.row_val_.count_ = column_num;
sparse_row.column_ids_ = sparse_col_id_array;
sparse_row.capacity_ = column_num;
int64_t pos = 0;
int64_t rowkey_start_pos = 0;
int64_t rowkey_length = 0;
int64_t read_pos = 0;
char* buf = get_serialize_buf();
// write success
pos = 0;
rowkey_start_pos = 0;
rowkey_length = 0;
ASSERT_EQ(OB_SUCCESS, row_generate_.get_next_row(writer_row));
// set nop value
int nop_cell_cnt = 5;
int nop_indexs[] = {5, 9, 12, 18, 21};
for (int i = 0; i < nop_cell_cnt; ++i) {
writer_row.row_val_.cells_[nop_indexs[i]].set_nop_value();
}
ret = row_writer.write_flat_row(
rowkey_column_count, writer_row, buf, 2 * 1024 * 1024, pos, rowkey_start_pos, rowkey_length);
ASSERT_EQ(OB_SUCCESS, ret);
// init ObColumnMap
column_map_.reset();
ObArray<ObColDesc> columns;
ASSERT_EQ(OB_SUCCESS, row_generate_.get_schema().get_column_ids(columns));
ret = init_column_map(writer_row, columns);
ASSERT_EQ(OB_SUCCESS, ret);
// read success
read_pos = 0;
ret = flat_row_reader.read_row(buf, pos, read_pos, column_map_, allocator_, sparse_row, SPARSE_ROW_STORE);
ASSERT_EQ(OB_SUCCESS, ret);
ASSERT_TRUE(column_num - nop_cell_cnt == sparse_row.row_val_.count_);
// every obj should equal
int index = 0;
for (int64_t i = 0; i < column_num; ++i) {
if (!writer_row.row_val_.cells_[i].is_nop_value()) {
ASSERT_TRUE(writer_row.row_val_.cells_[i] == sparse_row.row_val_.cells_[index]);
ASSERT_TRUE(columns[i].col_id_ == sparse_row.column_ids_[index]);
index++;
}
}
int sparse_buf_size = 10 * 1024;
char sparse_buf[sparse_buf_size];
MEMSET(sparse_buf, 0, sparse_buf_size);
// write sparse row
pos = 0;
ret = row_writer.write_sparse_row(
rowkey_column_count, sparse_row, sparse_buf, sparse_buf_size, pos, rowkey_start_pos, rowkey_length);
ASSERT_EQ(OB_SUCCESS, ret);
ObStoreRow reader_row;
oceanbase::common::ObObj reader_objs[column_num];
reader_row.row_val_.cells_ = reader_objs;
reader_row.row_val_.count_ = column_num;
reader_row.capacity_ = column_num;
uint16_t col_id_array[column_num];
reader_row.column_ids_ = col_id_array;
// read success
read_pos = 0;
ret = sparse_row_reader.read_row(sparse_buf, pos, read_pos, column_map_, allocator_, reader_row, FLAT_ROW_STORE);
ASSERT_EQ(OB_SUCCESS, ret);
ASSERT_EQ(false, reader_row.is_sparse_row_);
ASSERT_TRUE(column_num == reader_row.row_val_.count_);
// every obj should equal
for (int64_t i = 0; i < column_num; ++i) {
if (!writer_row.row_val_.cells_[i].is_nop_value()) {
ASSERT_TRUE(writer_row.row_val_.cells_[i] == reader_row.row_val_.cells_[i]);
} else {
ASSERT_EQ(true, reader_row.row_val_.cells_[i].is_nop_value());
}
}
}
TEST_F(TestRowReader, read_sparse_row_from_sparse_storage)
{
int ret = OB_SUCCESS;
ObRowWriter row_writer;
oceanbase::common::ObObj objs[column_num];
ObStoreRow writer_row;
writer_row.row_val_.cells_ = objs;
writer_row.row_val_.count_ = column_num;
ObSparseRowReader row_reader;
ObStoreRow sparse_row;
oceanbase::common::ObObj sparse_row_objs[column_num];
sparse_row.row_val_.cells_ = sparse_row_objs;
sparse_row.row_val_.count_ = column_num;
uint16_t sparse_column_id[column_num];
sparse_row.column_ids_ = sparse_column_id;
sparse_row.capacity_ = column_num;
int64_t pos = 0;
int64_t rowkey_start_pos = 0;
int64_t rowkey_length = 0;
int64_t read_pos = 0;
char* buf = get_serialize_buf();
// write success
pos = 0;
rowkey_start_pos = 0;
rowkey_length = 0;
ASSERT_EQ(OB_SUCCESS, row_generate_.get_next_row(writer_row));
ObArray<ObColDesc> columns;
ASSERT_EQ(OB_SUCCESS, row_generate_.get_schema().get_column_ids(columns));
ObObjMeta meta_type_array[column_num];
ret = init_column_map(writer_row, columns);
ASSERT_EQ(OB_SUCCESS, ret);
// set nop value
int sparse_cell_cnt = 5;
int sparse_indexs[] = {5, 9, 12, 18, 21, INT_MAX};
for (int i = 0; i < sparse_cell_cnt; ++i) {
sparse_row.row_val_.cells_[i] = writer_row.row_val_.cells_[sparse_indexs[i]];
sparse_row.column_ids_[i] = columns[sparse_indexs[i]].col_id_;
meta_type_array[i] = columns[sparse_indexs[i]].col_type_;
}
sparse_row.row_val_.count_ = sparse_cell_cnt;
sparse_row.is_sparse_row_ = true;
sparse_row.flag_ = writer_row.flag_;
sparse_row.dml_ = writer_row.dml_;
ret = row_writer.write_sparse_row(
rowkey_column_count, sparse_row, buf, 2 * 1024 * 1024, pos, rowkey_start_pos, rowkey_length);
ASSERT_EQ(OB_SUCCESS, ret);
ObStoreRow reader_row;
oceanbase::common::ObObj reader_objs[column_num];
reader_row.row_val_.cells_ = reader_objs;
reader_row.row_val_.count_ = column_num;
reader_row.capacity_ = column_num;
ObStoreRow reader_flat_row;
oceanbase::common::ObObj reader_flat_objs[column_num];
reader_flat_row.row_val_.cells_ = reader_flat_objs;
reader_flat_row.row_val_.count_ = column_num;
reader_flat_row.capacity_ = column_num;
// read success
read_pos = 0;
ret = row_reader.read_full_row(buf, pos, read_pos, meta_type_array, allocator_, reader_row);
ASSERT_EQ(OB_SUCCESS, ret);
ASSERT_EQ(true, reader_row.is_sparse_row_);
ASSERT_TRUE(sparse_cell_cnt == reader_row.row_val_.count_);
// every obj should equal
int index = 0;
for (int64_t i = 0; i < reader_row.row_val_.count_; ++i) {
ASSERT_TRUE(sparse_row.row_val_.cells_[i] == reader_row.row_val_.cells_[i]);
ASSERT_TRUE(sparse_row.column_ids_[i] == reader_row.column_ids_[i]);
}
// read success
read_pos = 0;
ret = row_reader.read_row(buf, pos, read_pos, column_map_, allocator_, reader_flat_row, FLAT_ROW_STORE);
ASSERT_EQ(OB_SUCCESS, ret);
ASSERT_EQ(false, reader_flat_row.is_sparse_row_);
ASSERT_TRUE(column_num == reader_flat_row.row_val_.count_);
// every obj should equal
index = 0;
for (int64_t i = 0; i < column_num; ++i) {
if (i == sparse_indexs[index]) {
ASSERT_TRUE(writer_row.row_val_.cells_[i] == reader_flat_row.row_val_.cells_[i]);
index++;
} else {
ASSERT_EQ(true, reader_flat_row.row_val_.cells_[i].is_nop_value());
}
}
}
TEST_F(TestRowReader, sparse_row_test)
{
int ret = OB_SUCCESS;
ObRowWriter row_writer;
oceanbase::common::ObObj objs[column_num];
ObStoreRow writer_row;
writer_row.row_val_.cells_ = objs;
writer_row.row_val_.count_ = column_num;
ObSparseRowReader row_reader;
ObStoreRow sparse_row;
oceanbase::common::ObObj sparse_row_objs[column_num];
sparse_row.row_val_.cells_ = sparse_row_objs;
sparse_row.row_val_.count_ = column_num;
uint16_t column_ids[column_num];
sparse_row.column_ids_ = column_ids;
sparse_row.capacity_ = column_num;
int64_t pos = 0;
int64_t rowkey_start_pos = 0;
int64_t rowkey_length = 0;
int64_t read_pos = 0;
char* buf = get_serialize_buf();
// write success
pos = 0;
rowkey_start_pos = 0;
rowkey_length = 0;
ASSERT_EQ(OB_SUCCESS, row_generate_.get_next_row(writer_row));
ObArray<ObColDesc> columns;
ASSERT_EQ(OB_SUCCESS, row_generate_.get_schema().get_column_ids(columns));
ret = init_column_map(writer_row, columns);
ASSERT_EQ(OB_SUCCESS, ret);
// set nop value
int sparse_cell_cnt = 5;
int sparse_indexs[] = {5, 9, 12, 18, 21, INT_MAX};
for (int i = 0; i < sparse_cell_cnt; ++i) {
sparse_row.row_val_.cells_[i] = writer_row.row_val_.cells_[sparse_indexs[i]];
sparse_row.column_ids_[i] = columns[sparse_indexs[i]].col_id_;
}
sparse_row.row_val_.count_ = sparse_cell_cnt;
sparse_row.is_sparse_row_ = true;
sparse_row.flag_ = writer_row.flag_;
sparse_row.dml_ = writer_row.dml_;
int row_cnt = 100;
for (int i = 0; i < row_cnt; ++i) {
ret = row_writer.write_sparse_row(
rowkey_column_count, sparse_row, buf, 2 * 1024 * 1024, pos, rowkey_start_pos, rowkey_length);
ASSERT_EQ(OB_SUCCESS, ret);
}
ObStoreRow reader_row;
oceanbase::common::ObObj reader_objs[column_num];
reader_row.row_val_.cells_ = reader_objs;
reader_row.row_val_.count_ = column_num;
reader_row.capacity_ = column_num;
read_pos = 0;
for (int j = 0; j < row_cnt; ++j) {
// read success
ret = row_reader.read_row(buf, pos, read_pos, column_map_, allocator_, reader_row, SPARSE_ROW_STORE);
ASSERT_EQ(OB_SUCCESS, ret);
ASSERT_EQ(true, reader_row.is_sparse_row_);
ASSERT_TRUE(sparse_cell_cnt == reader_row.row_val_.count_);
// every obj should equal
int index = 0;
for (int64_t i = 0; i < reader_row.row_val_.count_; ++i) {
ASSERT_TRUE(sparse_row.row_val_.cells_[i] == reader_row.row_val_.cells_[i]);
ASSERT_TRUE(sparse_row.column_ids_[i] == reader_row.column_ids_[i]);
}
}
ObStoreRow reader_flat_row;
oceanbase::common::ObObj reader_flat_objs[column_num];
reader_flat_row.row_val_.cells_ = reader_flat_objs;
reader_flat_row.row_val_.count_ = column_num;
reader_flat_row.capacity_ = column_num;
// read success
read_pos = 0;
reader_row.row_val_.count_ = column_num;
for (int j = 0; j < row_cnt; ++j) {
ret = row_reader.read_row(buf, pos, read_pos, column_map_, allocator_, reader_flat_row, FLAT_ROW_STORE);
ASSERT_EQ(OB_SUCCESS, ret);
ASSERT_EQ(false, reader_flat_row.is_sparse_row_);
ASSERT_TRUE(column_num == reader_flat_row.row_val_.count_);
// every obj should equal
int index = 0;
for (int64_t i = 0; i < column_num; ++i) {
if (i == sparse_indexs[index]) {
ASSERT_TRUE(writer_row.row_val_.cells_[i] == reader_flat_row.row_val_.cells_[i]);
index++;
} else {
ASSERT_EQ(true, reader_flat_row.row_val_.cells_[i].is_nop_value());
// STORAGE_LOG(WARN, "~~~", K(i), K(index));
}
}
}
}
TEST_F(TestRowReader, sparse_row_with_trans_id)
{
int ret = OB_SUCCESS;
ObRowWriter row_writer;
oceanbase::common::ObObj objs[column_num];
ObStoreRow writer_row;
writer_row.row_val_.cells_ = objs;
writer_row.row_val_.count_ = column_num;
ObSparseRowReader row_reader;
ObStoreRow sparse_row;
oceanbase::common::ObObj sparse_row_objs[column_num];
sparse_row.row_val_.cells_ = sparse_row_objs;
sparse_row.row_val_.count_ = column_num;
uint16_t sparse_column_id[column_num];
sparse_row.column_ids_ = sparse_column_id;
sparse_row.capacity_ = column_num;
int64_t pos = 0;
int64_t rowkey_start_pos = 0;
int64_t rowkey_length = 0;
int64_t read_pos = 0;
char* buf = get_serialize_buf();
// write success
pos = 0;
rowkey_start_pos = 0;
rowkey_length = 0;
ASSERT_EQ(OB_SUCCESS, row_generate_.get_next_row(writer_row));
ObArray<ObColDesc> columns;
ASSERT_EQ(OB_SUCCESS, row_generate_.get_schema().get_column_ids(columns));
ret = init_column_map(writer_row, columns);
ASSERT_EQ(OB_SUCCESS, ret);
// set nop value
int sparse_cell_cnt = 5;
int sparse_indexs[] = {5, 9, 12, 18, 21, INT_MAX};
for (int i = 0; i < sparse_cell_cnt; ++i) {
sparse_row.row_val_.cells_[i] = writer_row.row_val_.cells_[sparse_indexs[i]];
sparse_row.column_ids_[i] = columns[sparse_indexs[i]].col_id_;
}
transaction::ObTransID trans_id(common::ObAddr(2, 3333));
transaction::ObTransID* trans_id_ptr = &trans_id;
sparse_row.row_val_.count_ = sparse_cell_cnt;
sparse_row.is_sparse_row_ = true;
sparse_row.flag_ = writer_row.flag_;
sparse_row.dml_ = writer_row.dml_;
sparse_row.trans_id_ptr_ = trans_id_ptr;
sparse_row.row_type_flag_.set_uncommitted_row(true);
ret = row_writer.write_sparse_row(
rowkey_column_count, sparse_row, buf, 2 * 1024 * 1024, pos, rowkey_start_pos, rowkey_length, false);
ASSERT_EQ(OB_SUCCESS, ret);
ObStoreRow reader_row;
oceanbase::common::ObObj reader_objs[column_num];
reader_row.row_val_.cells_ = reader_objs;
reader_row.row_val_.count_ = column_num;
reader_row.capacity_ = column_num;
ObStoreRow reader_flat_row;
oceanbase::common::ObObj reader_flat_objs[column_num];
reader_flat_row.row_val_.cells_ = reader_flat_objs;
reader_flat_row.row_val_.count_ = column_num;
reader_flat_row.capacity_ = column_num;
// read success
read_pos = 0;
ret = row_reader.read_row(buf, pos, read_pos, column_map_, allocator_, reader_row, SPARSE_ROW_STORE);
ASSERT_EQ(OB_SUCCESS, ret);
ASSERT_EQ(true, reader_row.is_sparse_row_);
ASSERT_TRUE(sparse_cell_cnt == reader_row.row_val_.count_);
// every obj should equal
int index = 0;
for (int64_t i = 0; i < reader_row.row_val_.count_; ++i) {
ASSERT_TRUE(sparse_row.row_val_.cells_[i] == reader_row.row_val_.cells_[i]);
ASSERT_TRUE(sparse_row.column_ids_[i] == reader_row.column_ids_[i]);
}
ASSERT_TRUE(*sparse_row.trans_id_ptr_ == trans_id);
}
TEST_F(TestRowReader, read_index_column_in_flat_storage)
{
int ret = OB_SUCCESS;
ObRowWriter row_writer;
oceanbase::common::ObObj objs[column_num];
ObStoreRow writer_row;
writer_row.row_val_.cells_ = objs;
writer_row.row_val_.count_ = column_num;
ObFlatRowReader row_reader;
ObStoreRow reader_row;
oceanbase::common::ObObj reader_objs[column_num];
reader_row.row_val_.cells_ = reader_objs;
reader_row.row_val_.count_ = column_num;
reader_row.capacity_ = column_num;
int64_t pos = 0;
int64_t rowkey_start_pos = 0;
int64_t rowkey_length = 0;
int64_t read_pos = 0;
char* buf = get_serialize_buf();
// write success
pos = 0;
rowkey_start_pos = 0;
rowkey_length = 0;
ASSERT_EQ(OB_SUCCESS, row_generate_.get_next_row(writer_row));
ret = row_writer.write_flat_row(
rowkey_column_count, writer_row, buf, 2 * 1024 * 1024, pos, rowkey_start_pos, rowkey_length);
ASSERT_EQ(OB_SUCCESS, ret);
// init ObColumnMap
column_map_.reset();
ObArray<ObColDesc> columns;
ASSERT_EQ(OB_SUCCESS, row_generate_.get_schema().get_column_ids(columns));
for (int64_t i = 0; i < column_num; ++i) {
ObObjType obj_type = writer_row.row_val_.cells_[i].get_type();
ObObjMeta obj_meta;
obj_meta.set_type(obj_type);
if (ObVarcharType == obj_type || ObCharType == obj_type) {
obj_meta.set_collation_type(CS_TYPE_UTF8MB4_GENERAL_CI);
obj_meta.set_collation_level(CS_LEVEL_IMPLICIT);
}
columns[i].col_type_ = obj_meta;
}
ret = row_reader.setup_row(buf, pos, 0, column_num);
ASSERT_EQ(OB_SUCCESS, ret);
int column_index_cnt = 5;
int column_indexs[] = {12, 5, 21, 18, 9};
for (int64_t i = 0; i < column_index_cnt; ++i) {
row_reader.read_column(
columns[column_indexs[i]].col_type_, allocator_, column_indexs[i], reader_row.row_val_.cells_[i]);
ASSERT_TRUE(writer_row.row_val_.cells_[column_indexs[i]] == reader_row.row_val_.cells_[i]);
}
}
TEST_F(TestRowReader, flat_row_with_trans_id)
{
int ret = OB_SUCCESS;
ObRowWriter row_writer;
oceanbase::common::ObObj objs[column_num];
ObStoreRow writer_row;
writer_row.row_val_.cells_ = objs;
writer_row.row_val_.count_ = column_num;
ObFlatRowReader row_reader;
int64_t pos = 0;
int64_t rowkey_start_pos = 0;
int64_t rowkey_length = 0;
int64_t read_pos = 0;
char* buf = get_serialize_buf();
// write success
pos = 0;
rowkey_start_pos = 0;
rowkey_length = 0;
ASSERT_EQ(OB_SUCCESS, row_generate_.get_next_row(writer_row));
ObArray<ObColDesc> columns;
ASSERT_EQ(OB_SUCCESS, row_generate_.get_schema().get_column_ids(columns));
ret = init_column_map(writer_row, columns);
ASSERT_EQ(OB_SUCCESS, ret);
transaction::ObTransID trans_id(common::ObAddr(2, 3333));
transaction::ObTransID* trans_id_ptr = &trans_id;
writer_row.row_type_flag_.set_uncommitted_row(true);
writer_row.trans_id_ptr_ = trans_id_ptr;
ret = row_writer.write(
rowkey_column_count, writer_row, buf, 2 * 1024 * 1024, pos, rowkey_start_pos, rowkey_length, false);
ASSERT_EQ(OB_SUCCESS, ret);
ObStoreRow reader_row;
oceanbase::common::ObObj reader_objs[column_num];
reader_row.row_val_.cells_ = reader_objs;
reader_row.row_val_.count_ = column_num;
reader_row.capacity_ = column_num;
transaction::ObTransID read_trans_id;
transaction::ObTransID* read_trans_id_ptr = &read_trans_id;
// read success
read_pos = 0;
reader_row.trans_id_ptr_ = read_trans_id_ptr;
ret = row_reader.read_row(buf, pos, read_pos, column_map_, allocator_, reader_row, FLAT_ROW_STORE);
ASSERT_EQ(OB_SUCCESS, ret);
ASSERT_EQ(false, reader_row.is_sparse_row_);
ASSERT_TRUE(column_num == reader_row.row_val_.count_);
// every obj should equal
int index = 0;
for (int64_t i = 0; i < reader_row.row_val_.count_; ++i) {
ASSERT_TRUE(writer_row.row_val_.cells_[i] == reader_row.row_val_.cells_[i]);
}
ASSERT_TRUE(read_trans_id == trans_id);
}
TEST_F(TestRowReader, read_full_sparse_row)
{
int ret = OB_SUCCESS;
ObRowWriter row_writer;
oceanbase::common::ObObj objs[column_num];
ObStoreRow writer_row;
writer_row.row_val_.cells_ = objs;
writer_row.row_val_.count_ = column_num;
ObSparseRowReader row_reader;
ObStoreRow reader_row;
oceanbase::common::ObObj reader_objs[column_num];
reader_row.row_val_.cells_ = reader_objs;
reader_row.row_val_.count_ = column_num;
reader_row.capacity_ = column_num;
ObStoreRow sparse_row;
oceanbase::common::ObObj sparse_row_objs[column_num];
sparse_row.row_val_.cells_ = sparse_row_objs;
sparse_row.row_val_.count_ = column_num;
uint16_t column_ids[column_num];
sparse_row.column_ids_ = column_ids;
sparse_row.capacity_ = column_num;
int64_t pos = 0;
int64_t rowkey_start_pos = 0;
int64_t rowkey_length = 0;
int64_t read_pos = 0;
char* buf = get_serialize_buf();
// write success
pos = 0;
rowkey_start_pos = 0;
rowkey_length = 0;
ASSERT_EQ(OB_SUCCESS, row_generate_.get_next_row(writer_row));
ObArray<ObColDesc> columns;
ASSERT_EQ(OB_SUCCESS, row_generate_.get_schema().get_column_ids(columns));
// set sparse cell value
int sparse_cell_cnt = 5;
int sparse_indexs[] = {5, 9, 12, 18, 21, INT_MAX};
for (int i = 0; i < sparse_cell_cnt; ++i) {
sparse_row.row_val_.cells_[i] = writer_row.row_val_.cells_[sparse_indexs[i]];
sparse_row.column_ids_[i] = columns[sparse_indexs[i]].col_id_;
}
sparse_row.row_val_.count_ = sparse_cell_cnt;
sparse_row.is_sparse_row_ = true;
sparse_row.flag_ = writer_row.flag_;
sparse_row.dml_ = writer_row.dml_;
int row_cnt = 100;
ret = row_writer.write_sparse_row(
rowkey_column_count, sparse_row, buf, 2 * 1024 * 1024, pos, rowkey_start_pos, rowkey_length);
ASSERT_EQ(OB_SUCCESS, ret);
// init ObColumnMap
column_map_.reset();
for (int64_t i = 0; i < column_num; ++i) {
ObObjType obj_type = writer_row.row_val_.cells_[i].get_type();
ObObjMeta obj_meta;
obj_meta.set_type(obj_type);
if (ObVarcharType == obj_type || ObCharType == obj_type) {
obj_meta.set_collation_type(CS_TYPE_UTF8MB4_GENERAL_CI);
obj_meta.set_collation_level(CS_LEVEL_IMPLICIT);
}
columns[i].col_type_ = obj_meta;
}
ret = row_reader.read_full_row(buf, pos, 0, nullptr, allocator_, reader_row);
ASSERT_EQ(OB_SUCCESS, ret);
for (int64_t i = 0; i < reader_row.row_val_.count_; ++i) {
ASSERT_TRUE(writer_row.row_val_.cells_[sparse_indexs[i]] == reader_row.row_val_.cells_[i]);
}
}
TEST_F(TestRowReader, read_index_column_in_sparse_storage)
{
int ret = OB_SUCCESS;
ObRowWriter row_writer;
oceanbase::common::ObObj objs[column_num];
ObStoreRow writer_row;
writer_row.row_val_.cells_ = objs;
writer_row.row_val_.count_ = column_num;
ObSparseRowReader row_reader;
ObStoreRow reader_row;
oceanbase::common::ObObj reader_objs[column_num];
reader_row.row_val_.cells_ = reader_objs;
reader_row.row_val_.count_ = column_num;
reader_row.capacity_ = column_num;
ObStoreRow sparse_row;
oceanbase::common::ObObj sparse_row_objs[column_num];
sparse_row.row_val_.cells_ = sparse_row_objs;
sparse_row.row_val_.count_ = column_num;
uint16_t column_ids[column_num];
sparse_row.column_ids_ = column_ids;
sparse_row.capacity_ = column_num;
int64_t pos = 0;
int64_t rowkey_start_pos = 0;
int64_t rowkey_length = 0;
int64_t read_pos = 0;
char* buf = get_serialize_buf();
// write success
pos = 0;
rowkey_start_pos = 0;
rowkey_length = 0;
ASSERT_EQ(OB_SUCCESS, row_generate_.get_next_row(writer_row));
ObArray<ObColDesc> columns;
ASSERT_EQ(OB_SUCCESS, row_generate_.get_schema().get_column_ids(columns));
// set sparse cell value
int sparse_cell_cnt = 5;
int sparse_indexs[] = {5, 9, 12, 18, 21, INT_MAX};
for (int i = 0; i < sparse_cell_cnt; ++i) {
sparse_row.row_val_.cells_[i] = writer_row.row_val_.cells_[sparse_indexs[i]];
sparse_row.column_ids_[i] = columns[sparse_indexs[i]].col_id_;
}
sparse_row.row_val_.count_ = sparse_cell_cnt;
sparse_row.is_sparse_row_ = true;
sparse_row.flag_ = writer_row.flag_;
sparse_row.dml_ = writer_row.dml_;
int row_cnt = 100;
ret = row_writer.write_sparse_row(
rowkey_column_count, sparse_row, buf, 2 * 1024 * 1024, pos, rowkey_start_pos, rowkey_length);
ASSERT_EQ(OB_SUCCESS, ret);
// init ObColumnMap
column_map_.reset();
for (int64_t i = 0; i < column_num; ++i) {
ObObjType obj_type = writer_row.row_val_.cells_[i].get_type();
ObObjMeta obj_meta;
obj_meta.set_type(obj_type);
if (ObVarcharType == obj_type || ObCharType == obj_type) {
obj_meta.set_collation_type(CS_TYPE_UTF8MB4_GENERAL_CI);
obj_meta.set_collation_level(CS_LEVEL_IMPLICIT);
}
columns[i].col_type_ = obj_meta;
}
ret = row_reader.setup_row(buf, pos, 0);
ASSERT_EQ(OB_SUCCESS, ret);
for (int64_t i = 0; i < sparse_cell_cnt; ++i) {
row_reader.read_column(columns[sparse_indexs[i]].col_type_, allocator_, i, reader_row.row_val_.cells_[i]);
ASSERT_TRUE(writer_row.row_val_.cells_[sparse_indexs[i]] == reader_row.row_val_.cells_[i]);
}
}
TEST_F(TestRowReader, test_skip_column)
{
int ret = OB_SUCCESS;
ObRowWriter row_writer;
oceanbase::common::ObObj objs[column_num];
ObStoreRow writer_row;
writer_row.row_val_.cells_ = objs;
writer_row.row_val_.count_ = column_num;
ObFlatRowReader row_reader;
ObStoreRow reader_row;
oceanbase::common::ObObj reader_objs[column_num];
reader_row.row_val_.cells_ = reader_objs;
reader_row.row_val_.count_ = column_num;
reader_row.capacity_ = column_num;
int64_t pos = 0;
int64_t rowkey_start_pos = 0;
int64_t rowkey_length = 0;
int64_t read_pos = 0;
char* buf = get_serialize_buf();
ObArray<ObColDesc> column_desc_array;
ASSERT_EQ(OB_SUCCESS, row_generate_.get_schema().get_column_ids(column_desc_array));
// write success
pos = 0;
rowkey_start_pos = 0;
rowkey_length = 0;
ASSERT_EQ(OB_SUCCESS, row_generate_.get_next_row(writer_row));
for (int64_t i = 0; i < column_num; ++i) {
ObObjType obj_type = writer_row.row_val_.cells_[i].get_type();
if (obj_type == ObMediumIntType) {
objs[i].set_mediumint(1000000);
} else if (obj_type == ObIntType) {
objs[i].set_int(1000000000000000000);
}
}
ret = row_writer.write_flat_row(
rowkey_column_count, writer_row, buf, 2 * 1024 * 1024, pos, rowkey_start_pos, rowkey_length);
ASSERT_EQ(OB_SUCCESS, ret);
// init ObColumnMap
column_map_.reset();
ObArray<ObColDesc> out_cols;
ObArray<int32_t> projector;
for (int64_t i = 0; i < column_num; ++i) {
ObObjType obj_type = writer_row.row_val_.cells_[i].get_type();
ObObjMeta obj_meta;
obj_meta.set_type(obj_type);
if (ObVarcharType == obj_type || ObCharType == obj_type || ObNCharType == obj_type || ObNVarchar2Type == obj_type ||
ObTinyTextType == obj_type || ObTextType == obj_type || ObMediumTextType == obj_type ||
ObLongTextType == obj_type) {
obj_meta.set_collation_type(CS_TYPE_UTF8MB4_GENERAL_CI);
obj_meta.set_collation_level(CS_LEVEL_IMPLICIT);
} else if (ObHexStringType == obj_type) {
obj_meta.set_collation_level(CS_LEVEL_IMPLICIT);
} else if (ObRawType == obj_type) {
obj_meta.set_collation_type(CS_TYPE_BINARY);
}
ObColDesc col_desc = column_desc_array[i];
col_desc.col_type_ = obj_meta;
ASSERT_EQ(OB_SUCCESS, out_cols.push_back(col_desc));
ASSERT_EQ(OB_SUCCESS, projector.push_back(static_cast<int32_t>(i)));
}
ASSERT_EQ(OB_SUCCESS,
column_map_.init(allocator_,
row_generate_.get_schema().get_schema_version(),
row_generate_.get_schema().get_rowkey_column_num(),
column_num,
out_cols,
nullptr,
&projector));
// read success
read_pos = 0;
ret = row_reader.read_row(buf, pos, read_pos, column_map_, allocator_, reader_row);
ASSERT_EQ(OB_SUCCESS, ret);
// read success
read_pos = 0;
ret = row_reader.read_row(buf, pos, read_pos, column_map_, allocator_, reader_row);
ASSERT_EQ(OB_SUCCESS, ret);
// skip column error
pos = 0;
rowkey_start_pos = 0;
rowkey_length = 0;
ObObjMeta obj_meta;
ASSERT_EQ(OB_SUCCESS, row_generate_.get_next_row(writer_row));
ret = row_writer.write(rowkey_column_count, writer_row, buf, 2 * 1024 * 1024, pos, rowkey_start_pos, rowkey_length);
ASSERT_EQ(OB_SUCCESS, ret);
// skip rowkey column error
column_map_.reset();
out_cols.reset();
projector.reset();
ObColDesc col_desc;
ObObjType obj_type = writer_row.row_val_.cells_[0].get_type();
obj_meta.set_type(obj_type);
col_desc = column_desc_array[0];
col_desc.col_type_ = obj_meta;
ASSERT_EQ(OB_SUCCESS, out_cols.push_back(col_desc));
ASSERT_EQ(OB_SUCCESS, projector.push_back(-1));
obj_type = writer_row.row_val_.cells_[1].get_type();
obj_meta.set_type(obj_type);
col_desc = column_desc_array[1];
col_desc.col_type_ = obj_meta;
ASSERT_EQ(OB_SUCCESS, out_cols.push_back(col_desc));
ASSERT_EQ(OB_SUCCESS, projector.push_back(1));
obj_type = writer_row.row_val_.cells_[2].get_type();
obj_meta.set_type(obj_type);
col_desc = column_desc_array[2];
col_desc.col_type_ = obj_meta;
ASSERT_EQ(OB_SUCCESS, out_cols.push_back(col_desc));
ASSERT_EQ(OB_SUCCESS, projector.push_back(2));
ASSERT_EQ(OB_SUCCESS,
column_map_.init(allocator_,
row_generate_.get_schema().get_schema_version(),
row_generate_.get_schema().get_rowkey_column_num(),
column_num,
out_cols,
nullptr,
&projector));
read_pos = 0;
ret = row_reader.read_row(buf, 8 + 4 + 2, read_pos, column_map_, allocator_, reader_row);
ASSERT_EQ(OB_SIZE_OVERFLOW, ret);
}
TEST_F(TestRowReader, read_multi)
{
int ret = OB_SUCCESS;
ObRowWriter row_writer;
oceanbase::common::ObObj objs[column_num];
ObStoreRow writer_row;
writer_row.row_val_.cells_ = objs;
writer_row.row_val_.count_ = column_num;
ObFlatRowReader row_reader;
ObStoreRow reader_row;
oceanbase::common::ObObj reader_objs[column_num];
reader_row.row_val_.cells_ = reader_objs;
reader_row.row_val_.count_ = column_num;
reader_row.capacity_ = column_num;
int64_t pos;
int64_t rowkey_start_pos = 0;
int64_t rowkey_length = 0;
int64_t sum_pos = 0;
char* buf = get_serialize_buf();
// write success
// pos = 0;
rowkey_start_pos = 0;
rowkey_length = 0;
ASSERT_EQ(OB_SUCCESS, row_generate_.get_next_row(writer_row));
ret =
row_writer.write(rowkey_column_count, writer_row, buf, 2 * 1024 * 1024, sum_pos, rowkey_start_pos, rowkey_length);
pos = sum_pos;
ASSERT_EQ(OB_SUCCESS, ret);
// init ObColumnMap
column_map_.reset();
ObArray<ObColDesc> columns;
ASSERT_EQ(OB_SUCCESS, row_generate_.get_schema().get_column_ids(columns));
ret = init_column_map(writer_row, columns);
ASSERT_EQ(OB_SUCCESS, ret);
// read success
ret = row_reader.read_row(buf, pos, 0, column_map_, allocator_, reader_row);
ASSERT_EQ(OB_SUCCESS, ret);
// every reader obj should equal
ASSERT_EQ(OB_SUCCESS, row_generate_.get_next_row(0, writer_row));
for (int64_t i = 0; i < column_num; ++i) {
ASSERT_TRUE(writer_row.row_val_.cells_[i] == reader_row.row_val_.cells_[i])
<< "\n i: " << i << "\n writer: " << to_cstring(writer_row.row_val_.cells_[i])
<< "\n reader: " << to_cstring(reader_row.row_val_.cells_[i]);
}
// read success
ret = row_reader.read_row(buf, pos, 0, column_map_, allocator_, reader_row);
ASSERT_EQ(OB_SUCCESS, ret);
// every reader obj should equal
ASSERT_EQ(OB_SUCCESS, row_generate_.get_next_row(0, writer_row));
for (int64_t i = 0; i < column_num; ++i) {
ASSERT_TRUE(writer_row.row_val_.cells_[i] == reader_row.row_val_.cells_[i])
<< "\n i: " << i << "\n writer: " << to_cstring(writer_row.row_val_.cells_[i])
<< "\n reader: " << to_cstring(reader_row.row_val_.cells_[i]);
}
}
TEST_F(TestRowReader, read_flat_rowkey)
{
int ret = OB_SUCCESS;
ObRowWriter row_writer;
oceanbase::common::ObObj objs[column_num];
ObStoreRow writer_row;
writer_row.row_val_.cells_ = objs;
writer_row.row_val_.count_ = column_num;
ObFlatRowReader row_reader;
ObStoreRow reader_row;
oceanbase::common::ObObj reader_objs[column_num];
reader_row.row_val_.cells_ = reader_objs;
reader_row.row_val_.count_ = 0;
reader_row.capacity_ = column_num;
int64_t pos = 0;
int64_t rowkey_start_pos = 0;
int64_t rowkey_length = 0;
int64_t read_pos = 0;
char* buf = get_serialize_buf();
ObArray<ObColDesc> column_desc_array;
ASSERT_EQ(OB_SUCCESS, row_generate_.get_schema().get_column_ids(column_desc_array));
for (int64_t i = 0; i < column_num; ++i) {
ObObjType obj_type = writer_row.row_val_.cells_[i].get_type();
ObObjMeta obj_meta;
obj_meta.set_type(obj_type);
if (ObVarcharType == obj_type || ObCharType == obj_type) {
obj_meta.set_collation_type(CS_TYPE_UTF8MB4_GENERAL_CI);
obj_meta.set_collation_level(CS_LEVEL_IMPLICIT);
}
column_desc_array[i].col_type_ = obj_meta;
}
ASSERT_EQ(OB_SUCCESS,
column_map_.init(allocator_,
row_generate_.get_schema().get_schema_version(),
row_generate_.get_schema().get_rowkey_column_num(),
column_num,
column_desc_array));
{
ObNewRow new_writer_row;
new_writer_row.count_ = column_num;
new_writer_row.cells_ = objs;
ObNewRow new_reader_row;
new_reader_row.count_ = 0;
new_reader_row.cells_ = reader_objs;
pos = 0;
ASSERT_EQ(OB_SUCCESS, row_generate_.get_next_row(writer_row));
ret = row_writer.write(new_writer_row, buf, 2 * 1024 * 1024, FLAT_ROW_STORE, pos);
ASSERT_EQ(OB_SUCCESS, ret);
// init column type array
ObObjMeta column_types[common::OB_ROW_MAX_COLUMNS_COUNT];
for (int64_t i = 0; i < column_num; ++i) {
ObObjType obj_type = new_writer_row.cells_[i].get_type();
ObObjMeta& obj_meta = column_types[i];
obj_meta.set_type(obj_type);
if (ObVarcharType == obj_type || ObCharType == obj_type) {
obj_meta.set_collation_type(CS_TYPE_UTF8MB4_GENERAL_CI);
obj_meta.set_collation_level(CS_LEVEL_IMPLICIT);
}
}
// buf is NULL
read_pos = 0;
ret = row_reader.read_compact_rowkey(NULL, 0, NULL, pos, read_pos, new_reader_row);
ASSERT_EQ(OB_INVALID_ARGUMENT, ret);
// read_sequence_columns overflow
ret = row_reader.read_compact_rowkey(
column_types, column_map_.get_rowkey_store_count(), buf, 1, read_pos, new_reader_row);
// FIXEME: we don't check this overflow type
ASSERT_EQ(OB_SUCCESS, ret);
// new row read success
read_pos = 0;
ret = row_reader.read_compact_rowkey(
column_types, column_map_.get_rowkey_store_count(), buf, pos, read_pos, new_reader_row);
ASSERT_EQ(OB_SUCCESS, ret);
// every reader obj should equal
for (int64_t i = 0; i < rowkey_column_count; ++i) {
ASSERT_TRUE(new_writer_row.cells_[i] == new_reader_row.cells_[i])
<< "\n i: " << i << "\n writer: " << to_cstring(new_writer_row.cells_[i])
<< "\n reader: " << to_cstring(new_reader_row.cells_[i]);
}
}
}
TEST_F(TestRowReader, read_sparse_rowkey)
{
int ret = OB_SUCCESS;
ObRowWriter row_writer;
oceanbase::common::ObObj objs[column_num];
ObStoreRow writer_row;
writer_row.row_val_.cells_ = objs;
writer_row.row_val_.count_ = column_num;
ObSparseRowReader row_reader;
ObStoreRow reader_row;
oceanbase::common::ObObj reader_objs[column_num];
reader_row.row_val_.cells_ = reader_objs;
reader_row.row_val_.count_ = 0;
reader_row.capacity_ = column_num;
int64_t pos = 0;
int64_t rowkey_start_pos = 0;
int64_t rowkey_length = 0;
int64_t read_pos = 0;
char* buf = get_serialize_buf();
ObArray<ObColDesc> column_desc_array;
ASSERT_EQ(OB_SUCCESS, row_generate_.get_schema().get_column_ids(column_desc_array));
for (int64_t i = 0; i < column_num; ++i) {
ObObjType obj_type = writer_row.row_val_.cells_[i].get_type();
ObObjMeta obj_meta;
obj_meta.set_type(obj_type);
if (ObVarcharType == obj_type || ObCharType == obj_type) {
obj_meta.set_collation_type(CS_TYPE_UTF8MB4_GENERAL_CI);
obj_meta.set_collation_level(CS_LEVEL_IMPLICIT);
}
column_desc_array[i].col_type_ = obj_meta;
}
ASSERT_EQ(OB_SUCCESS,
column_map_.init(allocator_,
row_generate_.get_schema().get_schema_version(),
row_generate_.get_schema().get_rowkey_column_num(),
column_num,
column_desc_array));
{
ObNewRow new_writer_row;
new_writer_row.count_ = column_num;
new_writer_row.cells_ = objs;
ObNewRow new_reader_row;
new_reader_row.count_ = 0;
new_reader_row.cells_ = reader_objs;
pos = 0;
ASSERT_EQ(OB_SUCCESS, row_generate_.get_next_row(writer_row));
ret = row_writer.write(new_writer_row, buf, 2 * 1024 * 1024, SPARSE_ROW_STORE, pos);
ASSERT_EQ(OB_SUCCESS, ret);
// init column type array
ObObjMeta column_types[common::OB_ROW_MAX_COLUMNS_COUNT];
for (int64_t i = 0; i < column_num; ++i) {
ObObjType obj_type = new_writer_row.cells_[i].get_type();
ObObjMeta& obj_meta = column_types[i];
obj_meta.set_type(obj_type);
if (ObVarcharType == obj_type || ObCharType == obj_type) {
obj_meta.set_collation_type(CS_TYPE_UTF8MB4_GENERAL_CI);
obj_meta.set_collation_level(CS_LEVEL_IMPLICIT);
}
}
// buf is NULL
read_pos = 0;
ret = row_reader.read_compact_rowkey(NULL, 0, NULL, pos, read_pos, new_reader_row);
ASSERT_EQ(OB_INVALID_ARGUMENT, ret);
// read_sequence_columns overflow
ret = row_reader.read_compact_rowkey(
column_types, column_map_.get_rowkey_store_count(), buf, 1, read_pos, new_reader_row);
ASSERT_EQ(OB_BUF_NOT_ENOUGH, ret);
// new row read success
read_pos = 0;
ret = row_reader.read_compact_rowkey(
column_types, column_map_.get_rowkey_store_count(), buf, pos, read_pos, new_reader_row);
ASSERT_EQ(OB_SUCCESS, ret);
// every reader obj should equal
for (int64_t i = 0; i < rowkey_column_count; ++i) {
ASSERT_TRUE(new_writer_row.cells_[i] == new_reader_row.cells_[i]);
}
}
}
TEST_F(TestRowReader, read_not_exist_column)
{
int ret = OB_SUCCESS;
ObRowWriter row_writer;
oceanbase::common::ObObj objs[column_num];
ObStoreRow writer_row;
writer_row.row_val_.cells_ = objs;
writer_row.row_val_.count_ = column_num;
ObFlatRowReader row_reader;
ObStoreRow reader_row;
oceanbase::common::ObObj reader_objs[column_num];
reader_row.row_val_.cells_ = reader_objs;
reader_row.row_val_.count_ = column_num;
reader_row.capacity_ = column_num;
int64_t pos = 0;
int64_t rowkey_start_pos = 0;
int64_t rowkey_length = 0;
int64_t read_pos = 0;
char* buf = get_serialize_buf();
ObArray<ObColDesc> column_desc_array;
ASSERT_EQ(OB_SUCCESS, row_generate_.get_schema().get_column_ids(column_desc_array));
// write success
pos = 0;
rowkey_start_pos = 0;
rowkey_length = 0;
ASSERT_EQ(OB_SUCCESS, row_generate_.get_next_row(writer_row));
ret = row_writer.write(rowkey_column_count, writer_row, buf, 2 * 1024 * 1024, pos, rowkey_start_pos, rowkey_length);
ASSERT_EQ(OB_SUCCESS, ret);
// init ObColumnMap
column_map_.reset();
ObArray<int32_t> projector;
for (int64_t i = 0; i < column_num; ++i) {
ObObjType obj_type = writer_row.row_val_.cells_[i].get_type();
ObObjMeta obj_meta;
obj_meta.set_type(obj_type);
if (ObVarcharType == obj_type || ObCharType == obj_type || ObNCharType == obj_type || ObNVarchar2Type == obj_type ||
ObTinyTextType == obj_type || ObTextType == obj_type || ObMediumTextType == obj_type ||
ObLongTextType == obj_type) {
obj_meta.set_collation_type(CS_TYPE_UTF8MB4_GENERAL_CI);
obj_meta.set_collation_level(CS_LEVEL_IMPLICIT);
}
column_desc_array[i].col_type_ = obj_meta;
if (i % 2 == 0 && i > rowkey_column_count) {
ASSERT_EQ(OB_SUCCESS, projector.push_back(static_cast<int32_t>(-1)));
} else {
ASSERT_EQ(OB_SUCCESS, projector.push_back(static_cast<int32_t>(i)));
}
}
ASSERT_EQ(OB_SUCCESS,
column_map_.init(allocator_,
row_generate_.get_schema().get_schema_version(),
row_generate_.get_schema().get_rowkey_column_num(),
column_num,
column_desc_array,
nullptr,
&projector));
// read success
read_pos = 0;
ret = row_reader.read_row(buf, pos, read_pos, column_map_, allocator_, reader_row);
ASSERT_EQ(OB_SUCCESS, ret);
// every reader obj should equal
for (int64_t i = 0; i < column_num; ++i) {
if (i % 2 == 0 && i > rowkey_column_count) {
ASSERT_TRUE(reader_row.row_val_.cells_[i].is_nop_value());
} else {
ASSERT_TRUE(writer_row.row_val_.cells_[i] == reader_row.row_val_.cells_[i]);
}
}
}
TEST_F(TestRowReader, unknown_type)
{
int ret = OB_SUCCESS;
ObRowWriter row_writer;
oceanbase::common::ObObj objs[column_num];
ObStoreRow writer_row;
writer_row.row_val_.cells_ = objs;
writer_row.row_val_.count_ = column_num;
ObFlatRowReader row_reader;
ObStoreRow reader_row;
oceanbase::common::ObObj reader_objs[column_num];
reader_row.row_val_.cells_ = reader_objs;
reader_row.row_val_.count_ = column_num;
reader_row.capacity_ = column_num;
int64_t pos = 0;
int64_t rowkey_start_pos = 0;
int64_t rowkey_length = 0;
int64_t read_pos = 0;
char* buf = get_serialize_buf();
// write success
pos = 0;
rowkey_start_pos = 0;
rowkey_length = 0;
ASSERT_EQ(OB_SUCCESS, row_generate_.get_next_row(writer_row));
ret = row_writer.write(rowkey_column_count, writer_row, buf, 2 * 1024 * 1024, pos, rowkey_start_pos, rowkey_length);
ASSERT_EQ(OB_SUCCESS, ret);
// init ObColumnMap
column_map_.reset();
ObArray<ObColDesc> column_desc_array;
ASSERT_EQ(OB_SUCCESS, row_generate_.get_schema().get_column_ids(column_desc_array));
ObArray<ObColDesc> out_cols;
ObArray<int32_t> projector;
for (int64_t i = 0; i < column_num;) {
ObObjType obj_type = writer_row.row_val_.cells_[i].get_type();
ObObjMeta obj_meta;
obj_meta.set_type(obj_type);
if (ObVarcharType == obj_type || ObCharType == obj_type || ObNCharType == obj_type || ObNVarchar2Type == obj_type ||
ObTinyTextType == obj_type || ObTextType == obj_type || ObMediumTextType == obj_type ||
ObLongTextType == obj_type) {
obj_meta.set_collation_type(CS_TYPE_UTF8MB4_GENERAL_CI);
obj_meta.set_collation_level(CS_LEVEL_IMPLICIT);
}
if (2 == i) {
obj_meta.set_type(static_cast<ObObjType>(ObUnknownType));
obj_meta.set_collation_type(CS_TYPE_UTF8MB4_GENERAL_CI);
obj_meta.set_collation_level(CS_LEVEL_IMPLICIT);
}
ObColDesc col_desc = column_desc_array[i];
col_desc.col_type_ = obj_meta;
ASSERT_EQ(OB_SUCCESS, out_cols.push_back(col_desc));
ASSERT_EQ(OB_SUCCESS, projector.push_back(static_cast<int32_t>(i)));
i += 2;
}
ASSERT_EQ(OB_SUCCESS,
column_map_.init(allocator_,
row_generate_.get_schema().get_schema_version(),
row_generate_.get_schema().get_rowkey_column_num(),
column_num,
out_cols,
nullptr,
&projector));
read_pos = 0;
ret = row_reader.read_row(buf, pos, read_pos, column_map_, allocator_, reader_row);
ASSERT_EQ(OB_NOT_SUPPORTED, ret);
}
TEST_F(TestRowReader, obj_cast_write_read_different)
{
int ret = OB_SUCCESS;
ObRowWriter row_writer;
oceanbase::common::ObObj objs[column_num];
ObStoreRow writer_row;
writer_row.row_val_.cells_ = objs;
writer_row.row_val_.count_ = column_num;
ObFlatRowReader row_reader;
ObStoreRow reader_row;
oceanbase::common::ObObj reader_objs[column_num];
reader_row.row_val_.cells_ = reader_objs;
reader_row.row_val_.count_ = column_num;
reader_row.capacity_ = column_num;
int64_t pos = 0;
int64_t rowkey_start_pos = 0;
int64_t rowkey_length = 0;
int64_t read_pos = 0;
// write success
char* buf = get_serialize_buf();
pos = 0;
rowkey_start_pos = 0;
rowkey_length = 0;
ASSERT_EQ(OB_SUCCESS, row_generate_.get_next_row(writer_row));
ret = row_writer.write(rowkey_column_count, writer_row, buf, 2 * 1024 * 1024, pos, rowkey_start_pos, rowkey_length);
ASSERT_EQ(OB_SUCCESS, ret);
// every type cast to others
for (int64_t i = 0; i < column_num; ++i) {
// not support cast
for (int64_t j = 0; j < column_num; ++j) {
if (!cast_supported(
objs[j].get_type(), objs[j].get_collation_type(), objs[i].get_type(), objs[i].get_collation_type())) {
ret = OB_NOT_SUPPORTED;
}
}
// init ObColumnMap
column_map_.reset();
ObArray<ObColDesc> columns;
ASSERT_EQ(OB_SUCCESS, row_generate_.get_schema().get_column_ids(columns));
ret = init_column_map(writer_row, columns);
ASSERT_EQ(OB_SUCCESS, ret);
// read success
read_pos = 0;
ret = row_reader.read_row(buf, pos, read_pos, column_map_, allocator_, reader_row);
if (OB_SUCCESS != ret && OB_NOT_SUPPORTED != ret && OB_DATA_OUT_OF_RANGE != ret && OB_INVALID_DATE_VALUE != ret) {
ASSERT_TRUE(false);
}
}
}
TEST_F(TestRowReader, sparse_row_about_lob)
{
int ret = OB_SUCCESS;
ObRowWriter row_writer;
oceanbase::common::ObObj objs[column_num];
ObStoreRow writer_row;
writer_row.row_val_.cells_ = objs;
writer_row.row_val_.count_ = column_num;
ObSparseRowReader row_reader;
ObStoreRow sparse_row;
oceanbase::common::ObObj sparse_row_objs[column_num];
sparse_row.row_val_.cells_ = sparse_row_objs;
sparse_row.row_val_.count_ = column_num;
uint16_t column_ids[column_num];
sparse_row.column_ids_ = column_ids;
sparse_row.capacity_ = column_num;
int64_t pos = 0;
int64_t rowkey_start_pos = 0;
int64_t rowkey_length = 0;
int64_t read_pos = 0;
char* buf = get_serialize_buf();
// write success
pos = 0;
rowkey_start_pos = 0;
rowkey_length = 0;
ASSERT_EQ(OB_SUCCESS, row_generate_.get_next_row(writer_row));
ObArray<ObColDesc> columns;
ASSERT_EQ(OB_SUCCESS, row_generate_.get_schema().get_column_ids(columns));
// set nop value
int sparse_cell_cnt = 5;
int sparse_indexs[] = {5, 9, 12, 18, 21};
for (int i = 0; i < sparse_cell_cnt; ++i) {
sparse_row.row_val_.cells_[i] = writer_row.row_val_.cells_[sparse_indexs[i]];
sparse_row.column_ids_[i] = columns[sparse_indexs[i]].col_id_;
}
int64_t lob_buf_size = 1 * 1024 * 1024;
char lob_buf[lob_buf_size];
MEMSET(lob_buf, 'a', lob_buf_size);
ObString str(lob_buf);
ObScale scale;
scale = ObLobScale::STORE_IN_ROW;
uint64_t lob_column_id = 39;
uint64_t lob_column_idx = 20;
sparse_row.row_val_.cells_[sparse_cell_cnt].set_string(ObLongTextType, str);
sparse_row.row_val_.cells_[sparse_cell_cnt].set_collation_type(CS_TYPE_UTF8MB4_GENERAL_CI);
sparse_row.row_val_.cells_[sparse_cell_cnt].set_scale(scale);
sparse_row.column_ids_[sparse_cell_cnt] = lob_column_id;
for (int64_t i = 0; i < column_num; ++i) {
ObObjMeta obj_meta;
ObObjType obj_type = writer_row.row_val_.cells_[i].get_type();
obj_meta.set_type(obj_type);
if (ObVarcharType == obj_type || ObCharType == obj_type || ObLongTextType == obj_type) {
obj_meta.set_collation_type(CS_TYPE_UTF8MB4_GENERAL_CI);
obj_meta.set_collation_level(CS_LEVEL_IMPLICIT);
}
columns[i].col_type_ = obj_meta;
}
ASSERT_EQ(OB_SUCCESS,
column_map_.init(allocator_,
row_generate_.get_schema().get_schema_version(),
row_generate_.get_schema().get_rowkey_column_num(),
column_num,
columns));
sparse_cell_cnt++;
sparse_row.row_val_.count_ = sparse_cell_cnt;
sparse_row.is_sparse_row_ = true;
sparse_row.flag_ = writer_row.flag_;
sparse_row.dml_ = writer_row.dml_;
ret = row_writer.write_sparse_row(
rowkey_column_count, sparse_row, buf, 2 * 1024 * 1024, pos, rowkey_start_pos, rowkey_length);
ASSERT_EQ(OB_SUCCESS, ret);
ObStoreRow reader_row;
oceanbase::common::ObObj reader_objs[column_num];
reader_row.row_val_.cells_ = reader_objs;
reader_row.row_val_.count_ = column_num;
reader_row.capacity_ = column_num;
read_pos = 0;
// read success
ret = row_reader.read_row(buf, pos, read_pos, column_map_, allocator_, reader_row, SPARSE_ROW_STORE);
ASSERT_EQ(OB_SUCCESS, ret);
ASSERT_EQ(true, reader_row.is_sparse_row_);
ASSERT_TRUE(sparse_cell_cnt == reader_row.row_val_.count_);
// every obj should equal
int index = 0;
for (int64_t i = 0; i < reader_row.row_val_.count_; ++i) {
STORAGE_LOG(INFO, "compare column", K(i), K(sparse_row.row_val_.cells_[i]), K(reader_row.row_val_.cells_[i]));
ASSERT_TRUE(sparse_row.row_val_.cells_[i] == reader_row.row_val_.cells_[i]);
ASSERT_TRUE(sparse_row.column_ids_[i] == reader_row.column_ids_[i]);
}
STORAGE_LOG(INFO, "success to read sparse row", K(reader_row));
}
} // end namespace unittest
} // end namespace oceanbase
int main(int argc, char** argv)
{
system("rm -f test_row_reader.log*");
oceanbase::common::ObLogger::get_logger().set_log_level("WARN");
OB_LOGGER.set_file_name("test_row_reader.log");
OB_LOGGER.set_log_level("INFO");
testing::InitGoogleTest(&argc, argv);
return RUN_ALL_TESTS();
}
Reference in New Issue View Git Blame Copy Permalink