Skip to content
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

[fix](SortMerger)add error throwing mechanism #34334

Closed
wants to merge 2 commits into from
Closed

[fix](SortMerger)add error throwing mechanism #34334

Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
4d35cad
[fix](SortMerger)add error throwing mechanism
yoko123yoko Apr 30, 2024
0a3112a
Merge branch 'apache:master' into fix_34333
TomaYoko Apr 30, 2024
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
189 changes: 98 additions & 91 deletions be/src/vec/runtime/vsorted_run_merger.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -98,120 +98,127 @@ Status VSortedRunMerger::prepare(const vector<BlockSupplier>& input_runs) {
}

Status VSortedRunMerger::get_next(Block* output_block, bool* eos) {
Copy link
Contributor

Choose a reason for hiding this comment

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

warning: function 'get_next' has cognitive complexity of 64 (threshold 50) [readability-function-cognitive-complexity]

Status VSortedRunMerger::get_next(Block* output_block, bool* eos) {
                         ^
Additional context

be/src/vec/runtime/vsorted_run_merger.cpp:106: +1, including nesting penalty of 0, nesting level increased to 1

        if (_pending_cursor != nullptr) {
        ^

be/src/vec/runtime/vsorted_run_merger.cpp:108: +2, including nesting penalty of 1, nesting level increased to 2

            if (has_next_block(cursor)) {
            ^

be/src/vec/runtime/vsorted_run_merger.cpp:114: +1, including nesting penalty of 0, nesting level increased to 1

        if (_priority_queue.empty()) {
        ^

be/src/vec/runtime/vsorted_run_merger.cpp:117: +1, nesting level increased to 1

        } else if (_priority_queue.size() == 1) {
               ^

be/src/vec/runtime/vsorted_run_merger.cpp:119: +2, including nesting penalty of 1, nesting level increased to 2

            while (_offset != 0 && current->block_ptr() != nullptr) {
            ^

be/src/vec/runtime/vsorted_run_merger.cpp:119: +1

            while (_offset != 0 && current->block_ptr() != nullptr) {
                                ^

be/src/vec/runtime/vsorted_run_merger.cpp:120: +3, including nesting penalty of 2, nesting level increased to 3

                if (_offset >= current->rows - current->pos) {
                ^

be/src/vec/runtime/vsorted_run_merger.cpp:122: +4, including nesting penalty of 3, nesting level increased to 4

                    if (_pipeline_engine_enabled) {
                    ^

be/src/vec/runtime/vsorted_run_merger.cpp:128: +1, nesting level increased to 3

                } else {
                  ^

be/src/vec/runtime/vsorted_run_merger.cpp:134: +2, including nesting penalty of 1, nesting level increased to 2

            if (current->isFirst()) {
            ^

be/src/vec/runtime/vsorted_run_merger.cpp:135: +3, including nesting penalty of 2, nesting level increased to 3

                if (current->block_ptr() != nullptr) {
                ^

be/src/vec/runtime/vsorted_run_merger.cpp:137: +4, including nesting penalty of 3, nesting level increased to 4

                    if (_pipeline_engine_enabled) {
                    ^

be/src/vec/runtime/vsorted_run_merger.cpp:143: +1, nesting level increased to 3

                } else {
                  ^

be/src/vec/runtime/vsorted_run_merger.cpp:146: +1, nesting level increased to 2

            } else {
              ^

be/src/vec/runtime/vsorted_run_merger.cpp:147: +3, including nesting penalty of 2, nesting level increased to 3

                if (current->block_ptr() != nullptr) {
                ^

be/src/vec/runtime/vsorted_run_merger.cpp:148: +4, including nesting penalty of 3, nesting level increased to 4

                    for (int i = 0; i < current->all_columns.size(); i++) {
                    ^

be/src/vec/runtime/vsorted_run_merger.cpp:155: +4, including nesting penalty of 3, nesting level increased to 4

                    if (_pipeline_engine_enabled) {
                    ^

be/src/vec/runtime/vsorted_run_merger.cpp:161: +1, nesting level increased to 3

                } else {
                  ^

be/src/vec/runtime/vsorted_run_merger.cpp:165: +1, nesting level increased to 1

        } else {
          ^

be/src/vec/runtime/vsorted_run_merger.cpp:172: +2, including nesting penalty of 1, nesting level increased to 2

            if (num_columns != merged_columns.size()) {
            ^

be/src/vec/runtime/vsorted_run_merger.cpp:181: +2, including nesting penalty of 1, nesting level increased to 2

            while (!_priority_queue.empty()) {
            ^

be/src/vec/runtime/vsorted_run_merger.cpp:185: +3, including nesting penalty of 2, nesting level increased to 3

                if (_offset > 0) {
                ^

be/src/vec/runtime/vsorted_run_merger.cpp:187: +1, nesting level increased to 3

                } else {
                  ^

be/src/vec/runtime/vsorted_run_merger.cpp:188: +4, including nesting penalty of 3, nesting level increased to 4

                    for (size_t i = 0; i < num_columns; ++i) {
                    ^

be/src/vec/runtime/vsorted_run_merger.cpp:197: +3, including nesting penalty of 2, nesting level increased to 3

                if (!next_heap(current)) {
                ^

be/src/vec/runtime/vsorted_run_merger.cpp:201: +3, including nesting penalty of 2, nesting level increased to 3

                if (merged_rows == _batch_size) {
                ^

be/src/vec/runtime/vsorted_run_merger.cpp:207: +2, including nesting penalty of 1, nesting level increased to 2

            if (merged_rows == 0) {
            ^

be/src/vec/runtime/vsorted_run_merger.cpp:214: +1, including nesting penalty of 0, nesting level increased to 1

        if (_limit != -1 && _num_rows_returned >= _limit) {
        ^

be/src/vec/runtime/vsorted_run_merger.cpp:214: +1

        if (_limit != -1 && _num_rows_returned >= _limit) {
                         ^

be/src/vec/runtime/vsorted_run_merger.cpp:220: +1, including nesting penalty of 0, nesting level increased to 1

    } catch (const std::runtime_error& e) {
      ^

be/src/vec/runtime/vsorted_run_merger.cpp:222: +1, including nesting penalty of 0, nesting level increased to 1

    } catch (...) {
      ^

Copy link
Contributor

Choose a reason for hiding this comment

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

warning: function 'get_next' has cognitive complexity of 64 (threshold 50) [readability-function-cognitive-complexity]

Status VSortedRunMerger::get_next(Block* output_block, bool* eos) {
                         ^
Additional context

be/src/vec/runtime/vsorted_run_merger.cpp:106: +1, including nesting penalty of 0, nesting level increased to 1

        if (_pending_cursor != nullptr) {
        ^

be/src/vec/runtime/vsorted_run_merger.cpp:108: +2, including nesting penalty of 1, nesting level increased to 2

            if (has_next_block(cursor)) {
            ^

be/src/vec/runtime/vsorted_run_merger.cpp:114: +1, including nesting penalty of 0, nesting level increased to 1

        if (_priority_queue.empty()) {
        ^

be/src/vec/runtime/vsorted_run_merger.cpp:117: +1, nesting level increased to 1

        } else if (_priority_queue.size() == 1) {
               ^

be/src/vec/runtime/vsorted_run_merger.cpp:119: +2, including nesting penalty of 1, nesting level increased to 2

            while (_offset != 0 && current->block_ptr() != nullptr) {
            ^

be/src/vec/runtime/vsorted_run_merger.cpp:119: +1

            while (_offset != 0 && current->block_ptr() != nullptr) {
                                ^

be/src/vec/runtime/vsorted_run_merger.cpp:120: +3, including nesting penalty of 2, nesting level increased to 3

                if (_offset >= current->rows - current->pos) {
                ^

be/src/vec/runtime/vsorted_run_merger.cpp:122: +4, including nesting penalty of 3, nesting level increased to 4

                    if (_pipeline_engine_enabled) {
                    ^

be/src/vec/runtime/vsorted_run_merger.cpp:128: +1, nesting level increased to 3

                } else {
                  ^

be/src/vec/runtime/vsorted_run_merger.cpp:134: +2, including nesting penalty of 1, nesting level increased to 2

            if (current->isFirst()) {
            ^

be/src/vec/runtime/vsorted_run_merger.cpp:135: +3, including nesting penalty of 2, nesting level increased to 3

                if (current->block_ptr() != nullptr) {
                ^

be/src/vec/runtime/vsorted_run_merger.cpp:137: +4, including nesting penalty of 3, nesting level increased to 4

                    if (_pipeline_engine_enabled) {
                    ^

be/src/vec/runtime/vsorted_run_merger.cpp:143: +1, nesting level increased to 3

                } else {
                  ^

be/src/vec/runtime/vsorted_run_merger.cpp:146: +1, nesting level increased to 2

            } else {
              ^

be/src/vec/runtime/vsorted_run_merger.cpp:147: +3, including nesting penalty of 2, nesting level increased to 3

                if (current->block_ptr() != nullptr) {
                ^

be/src/vec/runtime/vsorted_run_merger.cpp:148: +4, including nesting penalty of 3, nesting level increased to 4

                    for (int i = 0; i < current->all_columns.size(); i++) {
                    ^

be/src/vec/runtime/vsorted_run_merger.cpp:154: +4, including nesting penalty of 3, nesting level increased to 4

                    if (_pipeline_engine_enabled) {
                    ^

be/src/vec/runtime/vsorted_run_merger.cpp:160: +1, nesting level increased to 3

                } else {
                  ^

be/src/vec/runtime/vsorted_run_merger.cpp:164: +1, nesting level increased to 1

        } else {
          ^

be/src/vec/runtime/vsorted_run_merger.cpp:170: +2, including nesting penalty of 1, nesting level increased to 2

            if (num_columns != merged_columns.size()) {
            ^

be/src/vec/runtime/vsorted_run_merger.cpp:179: +2, including nesting penalty of 1, nesting level increased to 2

            while (!_priority_queue.empty()) {
            ^

be/src/vec/runtime/vsorted_run_merger.cpp:183: +3, including nesting penalty of 2, nesting level increased to 3

                if (_offset > 0) {
                ^

be/src/vec/runtime/vsorted_run_merger.cpp:185: +1, nesting level increased to 3

                } else {
                  ^

be/src/vec/runtime/vsorted_run_merger.cpp:186: +4, including nesting penalty of 3, nesting level increased to 4

                    for (size_t i = 0; i < num_columns; ++i) {
                    ^

be/src/vec/runtime/vsorted_run_merger.cpp:194: +3, including nesting penalty of 2, nesting level increased to 3

                if (!next_heap(current)) {
                ^

be/src/vec/runtime/vsorted_run_merger.cpp:198: +3, including nesting penalty of 2, nesting level increased to 3

                if (merged_rows == _batch_size) {
                ^

be/src/vec/runtime/vsorted_run_merger.cpp:204: +2, including nesting penalty of 1, nesting level increased to 2

            if (merged_rows == 0) {
            ^

be/src/vec/runtime/vsorted_run_merger.cpp:211: +1, including nesting penalty of 0, nesting level increased to 1

        if (_limit != -1 && _num_rows_returned >= _limit) {
        ^

be/src/vec/runtime/vsorted_run_merger.cpp:211: +1

        if (_limit != -1 && _num_rows_returned >= _limit) {
                         ^

be/src/vec/runtime/vsorted_run_merger.cpp:216: +1, including nesting penalty of 0, nesting level increased to 1

    } catch (const std::runtime_error& e) {
      ^

be/src/vec/runtime/vsorted_run_merger.cpp:218: +1, including nesting penalty of 0, nesting level increased to 1

    } catch (...) {
      ^

ScopedTimer<MonotonicStopWatch> timer(_get_next_timer);
// Only have one receive data queue of data, no need to do merge and
// copy the data of block.
// return the data in receive data directly

if (_pending_cursor != nullptr) {
MergeSortCursor cursor(_pending_cursor);
if (has_next_block(cursor)) {
_priority_queue.push(cursor);
try {
ScopedTimer<MonotonicStopWatch> timer(_get_next_timer);
// Only have one receive data queue of data, no need to do merge and
// copy the data of block.
// return the data in receive data directly

if (_pending_cursor != nullptr) {
MergeSortCursor cursor(_pending_cursor);
if (has_next_block(cursor)) {
_priority_queue.push(cursor);
}
_pending_cursor = nullptr;
}
_pending_cursor = nullptr;
}

if (_priority_queue.empty()) {
*eos = true;
return Status::OK();
} else if (_priority_queue.size() == 1) {
auto current = _priority_queue.top();
while (_offset != 0 && current->block_ptr() != nullptr) {
if (_offset >= current->rows - current->pos) {
_offset -= (current->rows - current->pos);
if (_pipeline_engine_enabled) {
_pending_cursor = current.impl;
_priority_queue.pop();
return Status::OK();
if (_priority_queue.empty()) {
*eos = true;
return Status::OK();
} else if (_priority_queue.size() == 1) {
auto current = _priority_queue.top();
while (_offset != 0 && current->block_ptr() != nullptr) {
if (_offset >= current->rows - current->pos) {
_offset -= (current->rows - current->pos);
if (_pipeline_engine_enabled) {
_pending_cursor = current.impl;
_priority_queue.pop();
return Status::OK();
}
has_next_block(current);
} else {
current->pos += _offset;
_offset = 0;
}
has_next_block(current);
} else {
current->pos += _offset;
_offset = 0;
}
}

if (current->isFirst()) {
if (current->block_ptr() != nullptr) {
current->block_ptr()->swap(*output_block);
if (_pipeline_engine_enabled) {
_pending_cursor = current.impl;
_priority_queue.pop();
return Status::OK();
if (current->isFirst()) {
if (current->block_ptr() != nullptr) {
current->block_ptr()->swap(*output_block);
if (_pipeline_engine_enabled) {
_pending_cursor = current.impl;
_priority_queue.pop();
return Status::OK();
}
*eos = !has_next_block(current);
} else {
*eos = true;
}
*eos = !has_next_block(current);
} else {
*eos = true;
}
} else {
if (current->block_ptr() != nullptr) {
for (int i = 0; i < current->all_columns.size(); i++) {
auto& column_with_type = current->block_ptr()->get_by_position(i);
column_with_type.column = column_with_type.column->cut(
current->pos, current->rows - current->pos);
if (current->block_ptr() != nullptr) {
for (int i = 0; i < current->all_columns.size(); i++) {
auto& column_with_type = current->block_ptr()->get_by_position(i);
column_with_type.column = column_with_type.column->cut(
current->pos, current->rows - current->pos);
}
current->block_ptr()->swap(*output_block);
if (_pipeline_engine_enabled) {
_pending_cursor = current.impl;
_priority_queue.pop();
return Status::OK();
}
*eos = !has_next_block(current);
} else {
*eos = true;
}
current->block_ptr()->swap(*output_block);
if (_pipeline_engine_enabled) {
_pending_cursor = current.impl;
_priority_queue.pop();
return Status::OK();
}
*eos = !has_next_block(current);
} else {
*eos = true;
}
}
} else {
size_t num_columns = _empty_block.columns();
MutableBlock m_block =
} else {
size_t num_columns = _empty_block.columns();
MutableBlock m_block =
VectorizedUtils::build_mutable_mem_reuse_block(output_block, _empty_block);
MutableColumns& merged_columns = m_block.mutable_columns();
MutableColumns& merged_columns = m_block.mutable_columns();

if (num_columns != merged_columns.size()) {
throw Exception(
ErrorCode::INTERNAL_ERROR,
"num_columns!=merged_columns.size(), num_columns={}, merged_columns.size()={}",
num_columns, merged_columns.size());
}
if (num_columns != merged_columns.size()) {
throw Exception(ErrorCode::INTERNAL_ERROR,
"num_columns!=merged_columns.size(), "
"num_columns={}, merged_columns.size()={}",
num_columns, merged_columns.size());
}

/// Take rows from queue in right order and push to 'merged'.
size_t merged_rows = 0;
while (!_priority_queue.empty()) {
auto current = _priority_queue.top();
_priority_queue.pop();
/// Take rows from queue in right order and push to 'merged'.
size_t merged_rows = 0;
while (!_priority_queue.empty()) {
auto current = _priority_queue.top();
_priority_queue.pop();

if (_offset > 0) {
_offset--;
} else {
for (size_t i = 0; i < num_columns; ++i) {
merged_columns[i]->insert_from(*current->all_columns[i], current->pos);
}
++merged_rows;
}

if (_offset > 0) {
_offset--;
} else {
for (size_t i = 0; i < num_columns; ++i) {
merged_columns[i]->insert_from(*current->all_columns[i], current->pos);
// In pipeline engine, needs to check if the sender is readable
// before the next reading.
if (!next_heap(current)) {
return Status::OK();
}
++merged_rows;
}

// In pipeline engine, needs to check if the sender is readable before the next reading.
if (!next_heap(current)) {
return Status::OK();
if (merged_rows == _batch_size) {
break;
}
}
output_block->set_columns(std::move(merged_columns));

if (merged_rows == _batch_size) {
break;
if (merged_rows == 0) {
*eos = true;
return Status::OK();
}
}
output_block->set_columns(std::move(merged_columns));

if (merged_rows == 0) {
_num_rows_returned += output_block->rows();
if (_limit != -1 && _num_rows_returned >= _limit) {
output_block->set_num_rows(output_block->rows() - (_num_rows_returned - _limit));
*eos = true;
return Status::OK();
}
return Status::OK();
} catch (const std::runtime_error& e) {
return Status::RuntimeError(e.what());
} catch (...) {
throw;
}

_num_rows_returned += output_block->rows();
if (_limit != -1 && _num_rows_returned >= _limit) {
output_block->set_num_rows(output_block->rows() - (_num_rows_returned - _limit));
*eos = true;
}
return Status::OK();
}

bool VSortedRunMerger::next_heap(MergeSortCursor& current) {
Expand Down
Loading