Skip to content

[BUG] fix bug for vectorized compaction#9610

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
yiguolei merged 4 commits into from
May 19, 2022
Merged

[BUG] fix bug for vectorized compaction #9610

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
5bb3944
[BUG] fix bug for vectorized compaction
Gabriel39 May 17, 2022
400852d
update
Gabriel39 May 17, 2022
fc9dffd
update
Gabriel39 May 18, 2022
031d749
Update be/src/vec/olap/vcollect_iterator.cpp
Gabriel39 May 18, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
9 changes: 6 additions & 3 deletions be/src/olap/compaction.cpp
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -76,7 +76,10 @@ Status Compaction::do_compaction_impl(int64_t permits) {
_output_version =
Version(_input_rowsets.front()->start_version(), _input_rowsets.back()->end_version());

LOG(INFO) << "start " << compaction_name() << ". tablet=" << _tablet->full_name()
auto use_vectorized_compaction = _should_use_vectorized_compaction();
string merge_type = use_vectorized_compaction ? "v" : "";

LOG(INFO) << "start " << merge_type << compaction_name() << ". tablet=" << _tablet->full_name()
<< ", output_version=" << _output_version << ", permits: " << permits;

RETURN_NOT_OK(construct_output_rowset_writer());
Expand All@@ -87,15 +90,15 @@ Status Compaction::do_compaction_impl(int64_t permits) {
// The test results show that merger is low-memory-footprint, there is no need to tracker its mem pool
Merger::Statistics stats;
Status res;
auto use_vectorized_compaction = _should_use_vectorized_compaction();

if (use_vectorized_compaction) {
res = Merger::vmerge_rowsets(_tablet, compaction_type(), _input_rs_readers,
_output_rs_writer.get(), &stats);
} else {
res = Merger::merge_rowsets(_tablet, compaction_type(), _input_rs_readers,
_output_rs_writer.get(), &stats);
}
string merge_type = use_vectorized_compaction ? "v" : "";

if (!res.ok()) {
LOG(WARNING) << "fail to do " << merge_type << compaction_name() << ". res=" << res
<< ", tablet=" << _tablet->full_name()
Expand Down
2 changes: 1 addition & 1 deletion be/src/vec/olap/block_reader.cpp
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -64,7 +64,7 @@ Status BlockReader::_init_collect_iter(const ReaderParams& read_params,
}
}

_vcollect_iter.build_heap(*valid_rs_readers);
RETURN_IF_ERROR(_vcollect_iter.build_heap(*valid_rs_readers));
if (_vcollect_iter.is_merge()) {
auto status = _vcollect_iter.current_row(&_next_row);
_eof = status.precise_code() == OLAP_ERR_DATA_EOF;
Expand Down
39 changes: 31 additions & 8 deletions be/src/vec/olap/vcollect_iterator.cpp
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -26,6 +26,14 @@ namespace vectorized {

VCollectIterator::~VCollectIterator() {}

#define RETURN_IF_NOT_EOF_AND_OK(stmt) \
do { \
const Status& _status_ = (stmt); \
if (UNLIKELY(!_status_.ok() && _status_.precise_code() != OLAP_ERR_DATA_EOF)) { \
return _status_; \
} \
} while (false)

void VCollectIterator::init(TabletReader* reader) {
_reader = reader;
// when aggregate is enabled or key_type is DUP_KEYS, we don't merge
Expand All@@ -45,20 +53,24 @@ Status VCollectIterator::add_child(RowsetReaderSharedPtr rs_reader) {
// Build a merge heap. If _merge is true, a rowset with the max rownum
// status will be used as the base rowset, and the other rowsets will be merged first and
// then merged with the base rowset.
void VCollectIterator::build_heap(std::vector<RowsetReaderSharedPtr>& rs_readers) {
Status VCollectIterator::build_heap(std::vector<RowsetReaderSharedPtr>& rs_readers) {
DCHECK(rs_readers.size() == _children.size());
_skip_same = _reader->_tablet->tablet_schema().keys_type() == KeysType::UNIQUE_KEYS;
if (_children.empty()) {
_inner_iter.reset(nullptr);
return;
return Status::OK();
} else if (_merge) {
DCHECK(!rs_readers.empty());
for (auto [c_iter, r_iter] = std::pair {_children.begin(), rs_readers.begin()};
c_iter != _children.end();) {
if ((*c_iter)->init() != Status::OK()) {
auto s = (*c_iter)->init();
if (!s.ok()) {
delete (*c_iter);
Comment thread
Gabriel39 marked this conversation as resolved.
c_iter = _children.erase(c_iter);
r_iter = rs_readers.erase(r_iter);
if (s.precise_code() != OLAP_ERR_DATA_EOF) {
return s;
}
} else {
++c_iter;
++r_iter;
Expand DownExpand Up@@ -91,7 +103,7 @@ void VCollectIterator::build_heap(std::vector<RowsetReaderSharedPtr>& rs_readers
}
Level1Iterator* cumu_iter = new Level1Iterator(cumu_children, _reader,
cumu_children.size() > 1, _skip_same);
cumu_iter->init();
RETURN_IF_NOT_EOF_AND_OK(cumu_iter->init());
std::list<LevelIterator*> children;

@yiguoleiyiguoleiMay 18, 2022

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

cumu_iter->init(); called twice?

@Gabriel39Gabriel39May 18, 2022

Copy link
Copy Markdown
ContributorAuthor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Forgot to remove this line and I did this just now

children.push_back(*base_reader_child);
children.push_back(cumu_iter);
Expand All@@ -103,9 +115,10 @@ void VCollectIterator::build_heap(std::vector<RowsetReaderSharedPtr>& rs_readers
} else {
_inner_iter.reset(new Level1Iterator(_children, _reader, _merge, _skip_same));
}
_inner_iter->init();
RETURN_IF_NOT_EOF_AND_OK(_inner_iter->init());
// Clear _children earlier to release any related references
_children.clear();
return Status::OK();
}

bool VCollectIterator::LevelIteratorComparator::operator()(LevelIterator* lhs, LevelIterator* rhs) {
Expand DownExpand Up@@ -197,9 +210,13 @@ Status VCollectIterator::Level0Iterator::_refresh_current_row() {
_ref.row_pos = 0;
_block->clear_column_data();
auto res = _rs_reader->next_block(_block.get());
if (!res.ok()) {
if (!res.ok() && res.precise_code() != OLAP_ERR_DATA_EOF) {
return res;
}
if (res.precise_code() == OLAP_ERR_DATA_EOF && _block->rows() == 0) {
_ref.row_pos = -1;
return Status::OLAPInternalError(OLAP_ERR_DATA_EOF);
}
}
} while (_block->rows() != 0);
_ref.row_pos = -1;
Expand All@@ -209,7 +226,6 @@ Status VCollectIterator::Level0Iterator::_refresh_current_row() {
Status VCollectIterator::Level0Iterator::next(IteratorRowRef* ref) {
_ref.row_pos++;
RETURN_NOT_OK(_refresh_current_row());

*ref = _ref;
return Status::OK();
}
Expand All@@ -220,7 +236,14 @@ Status VCollectIterator::Level0Iterator::next(Block* block) {
_ref.row_pos = -1;
return Status::OK();
} else {
return _rs_reader->next_block(block);
auto res = _rs_reader->next_block(block);
if (!res.ok() && res.precise_code() != OLAP_ERR_DATA_EOF) {
return res;
}
if (res.precise_code() == OLAP_ERR_DATA_EOF && _block->rows() == 0) {
return Status::OLAPInternalError(OLAP_ERR_DATA_EOF);
}
return Status::OK();
}
}

Expand Down
2 changes: 1 addition & 1 deletion be/src/vec/olap/vcollect_iterator.h
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -49,7 +49,7 @@ class VCollectIterator {

Status add_child(RowsetReaderSharedPtr rs_reader);

void build_heap(std::vector<RowsetReaderSharedPtr>& rs_readers);
Status build_heap(std::vector<RowsetReaderSharedPtr>& rs_readers);
// Get top row of the heap, nullptr if reach end.
Status current_row(IteratorRowRef* ref) const;

Expand Down
5 changes: 4 additions & 1 deletion be/src/vec/olap/vgeneric_iterators.cpp
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -343,7 +343,10 @@ Status VMergeIterator::next_batch(vectorized::Block* block) {
delete ctx;
}
}

if (!_merge_heap.empty()) {
return Status::OK();
}
// Still last batch needs to be processed
return Status::EndOfFile("no more data in segment");
}

Expand Down