database/doris
2
0
Fork 0
Code Issues Pull Requests Actions 3 Packages Projects Releases Wiki Activity

[Memory Engine] Add MemTabletScan (#3734)

Browse Source
This commit is contained in:
Binglin Chang
2020-06-03 15:42:38 +08:00
committed by GitHub
parent ad7270b7ca
commit 70aa9d6ca8
20 changed files with 483 additions and 38 deletions
Download Patch File Download Diff File Expand all files Collapse all files

29
be/src/olap/base_tablet.cpp
View File

@ -16,22 +16,21 @@
// under the License.
#include "olap/base_tablet.h"
#include "util/path_util.h"
#include "olap/data_dir.h"
#include "util/path_util.h"
namespace doris {
BaseTablet::BaseTablet(TabletMetaSharedPtr tablet_meta, DataDir* data_dir) :
_state(tablet_meta->tablet_state()),
_tablet_meta(tablet_meta),
_schema(tablet_meta->tablet_schema()),
_data_dir(data_dir) {
BaseTablet::BaseTablet(TabletMetaSharedPtr tablet_meta, DataDir* data_dir)
: _state(tablet_meta->tablet_state()),
_tablet_meta(tablet_meta),
_schema(tablet_meta->tablet_schema()),
_data_dir(data_dir) {
_gen_tablet_path();
}
BaseTablet::~BaseTablet() {
}
BaseTablet::~BaseTablet() {}
OLAPStatus BaseTablet::set_tablet_state(TabletState state) {
if (_tablet_meta->tablet_state() == TABLET_SHUTDOWN && state != TABLET_SHUTDOWN) {
@ -44,11 +43,13 @@ OLAPStatus BaseTablet::set_tablet_state(TabletState state) {
}
void BaseTablet::_gen_tablet_path() {
std::string path = _data_dir->path() + DATA_PREFIX;
path = path_util::join_path_segments(path, std::to_string(_tablet_meta->shard_id()));
path = path_util::join_path_segments(path, std::to_string(_tablet_meta->tablet_id()));
path = path_util::join_path_segments(path, std::to_string(_tablet_meta->schema_hash()));
_tablet_path = path;
if (_data_dir != nullptr) {
std::string path = _data_dir->path() + DATA_PREFIX;
path = path_util::join_path_segments(path, std::to_string(_tablet_meta->shard_id()));
path = path_util::join_path_segments(path, std::to_string(_tablet_meta->tablet_id()));
path = path_util::join_path_segments(path, std::to_string(_tablet_meta->schema_hash()));
_tablet_path = path;
}
}
} /* namespace doris */

2
be/src/olap/memory/CMakeLists.txt
View File

@ -29,8 +29,10 @@ add_library(Memory STATIC
delta_index.cpp
hash_index.cpp
mem_tablet.cpp
mem_tablet_scan.cpp
mem_sub_tablet.cpp
partial_row_batch.cpp
row_block.cpp
schema.cpp
write_txn.cpp
)

1
be/src/olap/memory/column_block.cpp
View File

@ -64,6 +64,7 @@ Status ColumnBlock::copy_to(ColumnBlock* dest, size_t size, size_t esize) {
RETURN_IF_ERROR(dest->nulls().alloc(dest->size()));
memcpy(dest->nulls().data(), nulls().data(), size);
}
memcpy(dest->data().data(), data().data(), size * esize);
return Status::OK();
}

4
be/src/olap/memory/column_block.h
View File

@ -34,8 +34,12 @@ public:
Buffer& data() { return _data; }
const Buffer& data() const { return _data; }
Buffer& nulls() { return _nulls; }
const Buffer& nulls() const { return _nulls; }
// Allocate memory for this block, with space for size elements and each
// element have esize byte size
Status alloc(size_t size, size_t esize);

13
be/src/olap/memory/column_reader.h
View File

@ -36,6 +36,7 @@ namespace memory {
// Note: this class is only intended for single-thread single reader usage.
class ColumnBlockHolder {
public:
ColumnBlockHolder() {}
ColumnBlockHolder(ColumnBlock* cb, bool own) : _cb(cb), _own_cb(own) {}
void init(ColumnBlock* cb, bool own) {
@ -46,7 +47,9 @@ public:
void release() {
if (_own_cb) {
delete _cb;
// use delete _cb directly will cause DCHECK fail in DEBUG mode
// so we use scoped_refptr to free _cb instead
scoped_refptr<ColumnBlock> ref(_cb);
_own_cb = false;
}
_cb = nullptr;
@ -56,13 +59,7 @@ public:
bool own() const { return _own_cb; }
~ColumnBlockHolder() {
if (_own_cb) {
delete _cb;
_cb = nullptr;
_own_cb = false;
}
}
~ColumnBlockHolder() { release(); }
private:
ColumnBlock* _cb = nullptr;

6
be/src/olap/memory/mem_sub_tablet.cpp
View File

@ -78,7 +78,7 @@ Status MemSubTablet::read_column(uint64_t version, uint32_t cid,
cl = _columns[cid];
}
}
if (cl.get() != nullptr) {
if (cl.get() == nullptr) {
return Status::NotFound("column not found");
}
return cl->create_reader(version, reader);
@ -101,7 +101,7 @@ Status MemSubTablet::begin_write(scoped_refptr<Schema>* schema) {
// precache key columns
for (size_t i = 0; i < _schema->num_key_columns(); i++) {
uint32_t cid = _schema->get(i)->cid();
if (_writers[cid] != nullptr) {
if (_writers[cid] == nullptr) {
RETURN_IF_ERROR(_columns[cid]->create_writer(&_writers[cid]));
}
}
@ -202,7 +202,7 @@ Status MemSubTablet::commit_write(uint64_t version) {
}
{
std::lock_guard<std::mutex> lg(_lock);
if (_index != _write_index) {
if (_index.get() != _write_index.get()) {
_index = _write_index;
}
for (size_t cid = 0; cid < _writers.size(); cid++) {

34
be/src/olap/memory/mem_tablet.cpp
View File

@ -18,6 +18,7 @@
#include "olap/memory/mem_tablet.h"
#include "olap/memory/mem_sub_tablet.h"
#include "olap/memory/mem_tablet_scan.h"
#include "olap/memory/write_txn.h"
namespace doris {
@ -36,11 +37,35 @@ std::shared_ptr<MemTablet> MemTablet::create_tablet_from_meta(TabletMetaSharedPt
}
Status MemTablet::init() {
_max_version = 0;
return MemSubTablet::create(0, *_mem_schema.get(), &_sub_tablet);
}
Status MemTablet::scan(std::unique_ptr<ScanSpec>&& spec, std::unique_ptr<MemTabletScan>* scan) {
return Status::NotSupported("scan not supported");
Status MemTablet::scan(std::unique_ptr<ScanSpec>* spec, std::unique_ptr<MemTabletScan>* scan) {
uint64_t version = (*spec)->version();
if (version == UINT64_MAX) {
version = _max_version;
(*spec)->_version = version;
}
if (version > _max_version) {
return Status::InvalidArgument("Illegal scan version (larger than latest version)");
}
size_t num_rows = 0;
RETURN_IF_ERROR(_sub_tablet->get_size(version, &num_rows));
num_rows = std::min((*spec)->_limit, num_rows);
std::vector<std::unique_ptr<ColumnReader>> readers;
auto& columns = (*spec)->columns();
readers.resize(columns.size());
for (size_t i = 0; i < columns.size(); ++i) {
const ColumnSchema* cs = _mem_schema->get_by_name(columns[i]);
if (!cs) {
return Status::NotFound("column not found for scan");
}
RETURN_IF_ERROR(_sub_tablet->read_column(version, cs->cid(), &readers[i]));
}
scan->reset(new MemTabletScan(std::static_pointer_cast<MemTablet>(shared_from_this()), spec,
num_rows, &readers));
return Status::OK();
}
Status MemTablet::create_write_txn(std::unique_ptr<WriteTxn>* wtxn) {
@ -50,12 +75,15 @@ Status MemTablet::create_write_txn(std::unique_ptr<WriteTxn>* wtxn) {
Status MemTablet::commit_write_txn(WriteTxn* wtxn, uint64_t version) {
std::lock_guard<std::mutex> lg(_write_lock);
DCHECK_LT(_max_version, version);
RETURN_IF_ERROR(_sub_tablet->begin_write(&_mem_schema));
for (size_t i = 0; i < wtxn->batch_size(); i++) {
auto batch = wtxn->get_batch(i);
RETURN_IF_ERROR(_sub_tablet->apply_partial_row_batch(batch));
}
return _sub_tablet->commit_write(version);
RETURN_IF_ERROR(_sub_tablet->commit_write(version));
_max_version = version;
return Status::OK();
}
} // namespace memory

12
be/src/olap/memory/mem_tablet.h
View File

@ -52,12 +52,13 @@ public:
// Initialize
Status init();
// Scan the tablet, return a MemTabletScan object scan, user can specify projections,
// predicates and aggregations using ScanSpec, currently only support full scan with
// projection.
// Scan the tablet, return a MemTabletScan object scan, user can specify projections
// using ScanSpec, currently only support full scan with projection, will support
// filter/aggregation in the future.
//
// Note: spec will be passed to scan object
// Note: thread-safe, supports multi-reader concurrency.
Status scan(std::unique_ptr<ScanSpec>&& spec, std::unique_ptr<MemTabletScan>* scan);
Status scan(std::unique_ptr<ScanSpec>* spec, std::unique_ptr<MemTabletScan>* scan);
// Create a write transaction
//
@ -70,6 +71,7 @@ public:
Status commit_write_txn(WriteTxn* wtxn, uint64_t version);
private:
friend class MemTabletScan;
// memory::Schema is used internally rather than TabletSchema, so we need an extra
// copy of _schema with type memory::Schema.
scoped_refptr<Schema> _mem_schema;
@ -79,6 +81,8 @@ private:
std::mutex _write_lock;
std::atomic<uint64_t> _max_version;
DISALLOW_COPY_AND_ASSIGN(MemTablet);
};

59
be/src/olap/memory/mem_tablet_scan.cpp Normal file
View File

@ -0,0 +1,59 @@
// Licensed to the Apache Software Foundation (ASF) under one
// or more contributor license agreements. See the NOTICE file
// distributed with this work for additional information
// regarding copyright ownership. The ASF licenses this file
// to you under the Apache License, Version 2.0 (the
// "License"); you may not use this file except in compliance
// with the License. You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing,
// software distributed under the License is distributed on an
// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
// KIND, either express or implied. See the License for the
// specific language governing permissions and limitations
// under the License.
#include "olap/memory/mem_tablet_scan.h"
#include "olap/memory/column_reader.h"
#include "olap/memory/mem_sub_tablet.h"
#include "olap/memory/mem_tablet.h"
namespace doris {
namespace memory {
MemTabletScan::~MemTabletScan() {}
MemTabletScan::MemTabletScan(std::shared_ptr<MemTablet>&& tablet, std::unique_ptr<ScanSpec>* spec,
size_t num_rows, std::vector<std::unique_ptr<ColumnReader>>* readers)
: _tablet(std::move(tablet)),
_schema(_tablet->_mem_schema.get()),
_spec(std::move(*spec)),
_num_rows(num_rows),
_readers(std::move(*readers)) {
_row_block.reset(new RowBlock(_readers.size()));
_next_block = 0;
_num_blocks = num_block(_num_rows, Column::BLOCK_SIZE);
}
Status MemTabletScan::next_block(const RowBlock** block) {
if (_next_block >= _num_blocks) {
*block = nullptr;
return Status::OK();
}
size_t rows_in_block =
std::min((size_t)Column::BLOCK_SIZE, _num_rows - _next_block * Column::BLOCK_SIZE);
_row_block->_nrows = rows_in_block;
for (size_t i = 0; i < _readers.size(); ++i) {
RETURN_IF_ERROR(
_readers[i]->get_block(rows_in_block, _next_block, &_row_block->_columns[i]));
}
_next_block++;
*block = _row_block.get();
return Status::OK();
}
} // namespace memory
} // namespace doris

82
be/src/olap/memory/mem_tablet_scan.h Normal file
View File

@ -0,0 +1,82 @@
// Licensed to the Apache Software Foundation (ASF) under one
// or more contributor license agreements. See the NOTICE file
// distributed with this work for additional information
// regarding copyright ownership. The ASF licenses this file
// to you under the Apache License, Version 2.0 (the
// "License"); you may not use this file except in compliance
// with the License. You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing,
// software distributed under the License is distributed on an
// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
// KIND, either express or implied. See the License for the
// specific language governing permissions and limitations
// under the License.
#pragma once
#include "olap/memory/common.h"
#include "olap/memory/row_block.h"
#include "olap/memory/schema.h"
namespace doris {
namespace memory {
class ScanSpec {
public:
ScanSpec(vector<std::string>&& columns, uint64_t version = UINT64_MAX,
uint64_t limit = UINT64_MAX)
: _version(version), _limit(limit), _columns(std::move(columns)) {}
uint64_t version() const { return _version; }
uint64_t limit() const { return _limit; }
const vector<std::string> columns() const { return _columns; }
private:
friend class MemTablet;
uint64_t _version = UINT64_MAX;
uint64_t _limit = UINT64_MAX;
vector<std::string> _columns;
};
class HashIndex;
class MemTablet;
class ColumnReader;
// Scanner for MemTablet
class MemTabletScan {
public:
~MemTabletScan();
// Get next row_block, it will remain valid until next call to next_block.
Status next_block(const RowBlock** block);
private:
friend class MemTablet;
MemTabletScan(std::shared_ptr<MemTablet>&& tablet, std::unique_ptr<ScanSpec>* spec,
size_t num_rows, std::vector<std::unique_ptr<ColumnReader>>* readers);
std::shared_ptr<MemTablet> _tablet;
const Schema* _schema = nullptr;
std::unique_ptr<ScanSpec> _spec;
size_t _num_rows = 0;
size_t _num_blocks = 0;
// full scan support
std::vector<std::unique_ptr<ColumnReader>> _readers;
// returned block
std::unique_ptr<RowBlock> _row_block;
size_t _next_block = 0;
DISALLOW_COPY_AND_ASSIGN(MemTabletScan);
};
} // namespace memory
} // namespace doris

4
be/src/olap/memory/partial_row_batch.cpp
View File

@ -111,8 +111,8 @@ Status PartialRowBatch::cur_row_get_cell(size_t idx, const ColumnSchema** cs,
// Methods for PartialRowWriter
PartialRowWriter::PartialRowWriter(scoped_refptr<Schema>* schema)
: _schema(*schema), _bit_set_size(_schema->cid_size()), _bit_nullable_size(0) {
PartialRowWriter::PartialRowWriter(const scoped_refptr<Schema>& schema)
: _schema(schema), _bit_set_size(_schema->cid_size()), _bit_nullable_size(0) {
_temp_cells.resize(_schema->cid_size());
}

4
be/src/olap/memory/partial_row_batch.h
View File

@ -105,7 +105,7 @@ private:
// Example usage:
// scoped_refptr<Schema> sc;
// Schema::create("id int,uv int,pv int,city tinyint null", &sc);
// PartialRowWriter writer(*sc.get());
// PartialRowWriter writer(&sc);
// writer.start_batch();
// for (auto& row : rows) {
// writer.start_row();
@ -121,7 +121,7 @@ public:
static const size_t DEFAULT_BYTE_CAPACITY = 1 << 20;
static const size_t DEFAULT_ROW_CAPACIT = 1 << 16;
explicit PartialRowWriter(scoped_refptr<Schema>* schema);
explicit PartialRowWriter(const scoped_refptr<Schema>& schema);
~PartialRowWriter();
Status start_batch(size_t row_capacity = DEFAULT_ROW_CAPACIT,

28
be/src/olap/memory/row_block.cpp Normal file
View File

@ -0,0 +1,28 @@
// Licensed to the Apache Software Foundation (ASF) under one
// or more contributor license agreements. See the NOTICE file
// distributed with this work for additional information
// regarding copyright ownership. The ASF licenses this file
// to you under the Apache License, Version 2.0 (the
// "License"); you may not use this file except in compliance
// with the License. You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing,
// software distributed under the License is distributed on an
// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
// KIND, either express or implied. See the License for the
// specific language governing permissions and limitations
// under the License.
#include "olap/memory/row_block.h"
namespace doris {
namespace memory {
RowBlock::RowBlock(size_t num_columns) {
_columns.resize(num_columns);
}
} // namespace memory
} // namespace doris

42
be/src/olap/memory/row_block.h Normal file
View File

@ -0,0 +1,42 @@
// Licensed to the Apache Software Foundation (ASF) under one
// or more contributor license agreements. See the NOTICE file
// distributed with this work for additional information
// regarding copyright ownership. The ASF licenses this file
// to you under the Apache License, Version 2.0 (the
// "License"); you may not use this file except in compliance
// with the License. You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing,
// software distributed under the License is distributed on an
// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
// KIND, either express or implied. See the License for the
// specific language governing permissions and limitations
// under the License.
#pragma once
#include "olap/memory/column_block.h"
#include "olap/memory/column_reader.h"
#include "olap/memory/common.h"
namespace doris {
namespace memory {
class RowBlock {
public:
size_t num_rows() const { return _nrows; }
size_t num_columns() const { return _columns.size(); }
const ColumnBlock& get_column(size_t idx) const { return *_columns[idx].get(); }
private:
friend class MemTabletScan;
explicit RowBlock(size_t num_columns);
size_t _nrows = 0;
vector<ColumnBlockHolder> _columns;
};
} // namespace memory
} // namespace doris

2
be/src/olap/memory/schema.h
View File

@ -82,6 +82,8 @@ public:
inline size_t num_key_columns() const { return _tschema.num_key_columns(); }
const TabletSchema& get_tablet_schema() const { return _tschema; }
// Get ColumnSchema by index
const ColumnSchema* get(size_t idx) const;

2
be/src/olap/memory/typed_column_writer.h
View File

@ -201,7 +201,7 @@ public:
}
Status get_new_column(scoped_refptr<Column>* ret) {
if (*ret != _column) {
if (ret->get() != _column.get()) {
DLOG(INFO) << StringPrintf("%s switch new column", _column->debug_string().c_str());
(*ret).swap(_column);
_column.reset();

2
be/src/olap/memory/write_txn.h
View File

@ -40,6 +40,8 @@ public:
const Schema& schema() const { return *_schema.get(); }
scoped_refptr<Schema> get_schema_ptr() const { return _schema; }
// Get number of batches
size_t batch_size() const { return _batches.size(); }

1
be/test/olap/CMakeLists.txt
View File

@ -87,3 +87,4 @@ ADD_BE_TEST(memory/column_delta_test)
ADD_BE_TEST(memory/schema_test)
ADD_BE_TEST(memory/column_test)
ADD_BE_TEST(memory/partial_row_batch_test)
ADD_BE_TEST(memory/mem_tablet_test)

192
be/test/olap/memory/mem_tablet_test.cpp Normal file
View File

@ -0,0 +1,192 @@
// Licensed to the Apache Software Foundation (ASF) under one
// or more contributor license agreements. See the NOTICE file
// distributed with this work for additional information
// regarding copyright ownership. The ASF licenses this file
// to you under the Apache License, Version 2.0 (the
// "License"); you may not use this file except in compliance
// with the License. You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing,
// software distributed under the License is distributed on an
// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
// KIND, either express or implied. See the License for the
// specific language governing permissions and limitations
// under the License.
#include "olap/memory/mem_tablet.h"
#include <gtest/gtest.h>
#include "olap/memory/mem_tablet_scan.h"
#include "olap/memory/write_txn.h"
#include "olap/tablet_meta.h"
namespace doris {
namespace memory {
struct TData {
int32_t id;
int32_t uv;
int32_t pv;
int8_t city;
};
TEST(MemTablet, writescan) {
const int num_insert = 2000000;
const int insert_per_write = 500000;
const int num_update = 10000;
const int update_time = 3;
scoped_refptr<Schema> sc;
ASSERT_TRUE(Schema::create("id int,uv int,pv int,city tinyint null", &sc).ok());
std::unordered_map<uint32_t, uint32_t> col_idx_to_unique_id;
std::vector<TColumn> columns(sc->num_columns());
for (size_t i = 0; i < sc->num_columns(); i++) {
const ColumnSchema* cs = sc->get(i);
col_idx_to_unique_id[i] = cs->cid();
TColumn& c = columns[i];
c.__set_column_name(cs->name());
TColumnType tct;
if (cs->type() == ColumnType::OLAP_FIELD_TYPE_INT) {
tct.__set_type(TPrimitiveType::INT);
} else if (cs->type() == ColumnType::OLAP_FIELD_TYPE_TINYINT) {
tct.__set_type(TPrimitiveType::TINYINT);
} else {
ASSERT_TRUE(false);
}
c.__set_column_type(tct);
c.__set_is_allow_null(cs->is_nullable());
c.__set_is_key(cs->is_key());
c.__set_aggregation_type(TAggregationType::REPLACE);
}
TTabletSchema tschema;
tschema.__set_short_key_column_count(1);
tschema.__set_keys_type(TKeysType::UNIQUE_KEYS);
tschema.__set_columns(columns);
tschema.__set_is_in_memory(false);
tschema.__set_schema_hash(1);
TabletMetaSharedPtr tablet_meta(
new TabletMeta(1, 1, 1, 1, 1, tschema, static_cast<uint32_t>(sc->cid_size()),
col_idx_to_unique_id, TabletUid(1, 1), TTabletType::TABLET_TYPE_MEMORY));
std::shared_ptr<MemTablet> tablet = MemTablet::create_tablet_from_meta(tablet_meta, nullptr);
ASSERT_TRUE(tablet->init().ok());
uint64_t cur_version = 0;
vector<TData> alldata(num_insert);
// insert
srand(1);
size_t nrow = 0;
for (int insert_start = 0; insert_start < num_insert; insert_start += insert_per_write) {
std::unique_ptr<WriteTxn> wtx;
EXPECT_TRUE(tablet->create_write_txn(&wtx).ok());
PartialRowWriter writer(wtx->get_schema_ptr());
int insert_end = std::min(insert_start + insert_per_write, num_insert);
EXPECT_TRUE(writer.start_batch(insert_per_write + 1, insert_per_write * 32).ok());
for (int i = insert_start; i < insert_end; i++) {
nrow++;
EXPECT_TRUE(writer.start_row().ok());
int id = i;
int uv = rand() % 10000;
int pv = rand() % 10000;
int8_t city = rand() % 100;
alldata[i].id = id;
alldata[i].uv = uv;
alldata[i].pv = pv;
alldata[i].city = city;
EXPECT_TRUE(writer.set("id", &id).ok());
EXPECT_TRUE(writer.set("uv", &uv).ok());
EXPECT_TRUE(writer.set("pv", &pv).ok());
EXPECT_TRUE(writer.set("city", city % 2 == 0 ? nullptr : &city).ok());
EXPECT_TRUE(writer.end_row().ok());
}
vector<uint8_t> wtxn_buff;
EXPECT_TRUE(writer.finish_batch(&wtxn_buff).ok());
PartialRowBatch* batch = wtx->new_batch();
EXPECT_TRUE(batch->load(std::move(wtxn_buff)).ok());
EXPECT_TRUE(tablet->commit_write_txn(wtx.get(), ++cur_version).ok());
wtx.reset();
}
// update
for (int i = 0; i < update_time; i++) {
std::unique_ptr<WriteTxn> wtx;
EXPECT_TRUE(tablet->create_write_txn(&wtx).ok());
PartialRowWriter writer(wtx->get_schema_ptr());
EXPECT_TRUE(writer.start_batch(num_update + 1, num_update * 32).ok());
size_t nrow = 0;
for (int j = 0; j < num_update; j++) {
nrow++;
EXPECT_TRUE(writer.start_row().ok());
int id = rand() % num_insert;
int uv = rand() % 10000;
int pv = rand() % 10000;
int8_t city = rand() % 100;
alldata[id].uv = uv;
alldata[id].pv = pv;
alldata[id].city = city;
EXPECT_TRUE(writer.set("id", &id).ok());
EXPECT_TRUE(writer.set("pv", &pv).ok());
EXPECT_TRUE(writer.set("city", city % 2 == 0 ? nullptr : &city).ok());
EXPECT_TRUE(writer.end_row().ok());
}
vector<uint8_t> wtxn_buff;
EXPECT_TRUE(writer.finish_batch(&wtxn_buff).ok());
PartialRowBatch* batch = wtx->new_batch();
EXPECT_TRUE(batch->load(std::move(wtxn_buff)).ok());
EXPECT_TRUE(tablet->commit_write_txn(wtx.get(), ++cur_version).ok());
wtx.reset();
}
// scan perf test
{
double t0 = GetMonoTimeSecondsAsDouble();
std::unique_ptr<ScanSpec> scanspec(new ScanSpec({"pv"}, cur_version));
std::unique_ptr<MemTabletScan> scan;
ASSERT_TRUE(tablet->scan(&scanspec, &scan).ok());
const RowBlock* rblock = nullptr;
while (true) {
EXPECT_TRUE(scan->next_block(&rblock).ok());
if (!rblock) {
break;
}
}
double t = GetMonoTimeSecondsAsDouble() - t0;
LOG(INFO) << StringPrintf("scan %d record, time: %.3lfs %.0lf row/s", num_insert, t,
num_insert / t);
scan.reset();
}
// scan result validation
{
std::unique_ptr<ScanSpec> scanspec(new ScanSpec({"pv"}, cur_version));
std::unique_ptr<MemTabletScan> scan;
ASSERT_TRUE(tablet->scan(&scanspec, &scan).ok());
size_t curidx = 0;
while (true) {
const RowBlock* rblock = nullptr;
EXPECT_TRUE(scan->next_block(&rblock).ok());
if (!rblock) {
break;
}
size_t nrows = rblock->num_rows();
const ColumnBlock& cb = rblock->get_column(0);
for (size_t i = 0; i < nrows; i++) {
int32_t value = cb.data().as<int32_t>()[i];
EXPECT_EQ(value, alldata[curidx].pv);
curidx++;
}
}
EXPECT_EQ(curidx, (size_t)num_insert);
scan.reset();
}
}
} // namespace memory
} // namespace doris
int main(int argc, char** argv) {
testing::InitGoogleTest(&argc, argv);
return RUN_ALL_TESTS();
}

2
be/test/olap/memory/partial_row_batch_test.cpp
View File

@ -29,7 +29,7 @@ namespace memory {
TEST(PartialRowbatch, write) {
scoped_refptr<Schema> sc;
ASSERT_TRUE(Schema::create("id int,uv int,pv int,city tinyint null", &sc).ok());
PartialRowWriter writer(&sc);
PartialRowWriter writer(sc);
srand(1);
const int N = 1000;
size_t nrow = 0;