github-actions[bot] commented on code in PR #44469:
URL: https://github.com/apache/doris/pull/44469#discussion_r1879711280
##########
be/src/pipeline/exec/partitioned_aggregation_sink_operator.cpp:
##########
@@ -228,95 +245,105 @@ Status
PartitionedAggSinkLocalState::setup_in_memory_agg_op(RuntimeState* state)
return sink_local_state->open(state);
}
-Status PartitionedAggSinkLocalState::revoke_memory(RuntimeState* state) {
- VLOG_DEBUG << "query " << print_id(state->query_id()) << " agg node "
- << Base::_parent->node_id() << " revoke_memory"
+size_t PartitionedAggSinkOperatorX::get_reserve_mem_size(RuntimeState* state,
bool eos) {
+ auto& local_state = get_local_state(state);
+ auto* runtime_state = local_state._runtime_state.get();
+ auto size = _agg_sink_operator->get_reserve_mem_size(runtime_state, eos);
+ COUNTER_SET(local_state._memory_usage_reserved, int64_t(size));
+ return size;
+}
+
+Status PartitionedAggSinkLocalState::revoke_memory(
Review Comment:
warning: function 'revoke_memory' exceeds recommended size/complexity
thresholds [readability-function-size]
```cpp
Status PartitionedAggSinkLocalState::revoke_memory(
^
```
<details>
<summary>Additional context</summary>
**be/src/pipeline/exec/partitioned_aggregation_sink_operator.cpp:255:** 85
lines including whitespace and comments (threshold 80)
```cpp
Status PartitionedAggSinkLocalState::revoke_memory(
^
```
</details>
##########
be/src/pipeline/exec/spill_sort_sink_operator.cpp:
##########
@@ -164,88 +168,79 @@
if (eos) {
if (local_state._shared_state->is_spilled) {
if (revocable_mem_size(state) > 0) {
- RETURN_IF_ERROR(revoke_memory(state));
+ RETURN_IF_ERROR(revoke_memory(state, nullptr));
} else {
local_state._dependency->set_ready_to_read();
- local_state._finish_dependency->set_ready();
}
} else {
RETURN_IF_ERROR(
local_state._shared_state->in_mem_shared_state->sorter->prepare_for_read());
local_state._dependency->set_ready_to_read();
- local_state._finish_dependency->set_ready();
}
}
return Status::OK();
}
-Status SpillSortSinkLocalState::revoke_memory(RuntimeState* state) {
+size_t SpillSortSinkLocalState::get_reserve_mem_size(RuntimeState* state, bool
eos) {
+ auto& parent = Base::_parent->template cast<Parent>();
+ return
parent._sort_sink_operator->get_reserve_mem_size_for_next_sink(_runtime_state.get(),
+ eos);
+}
+
+Status SpillSortSinkLocalState::revoke_memory(RuntimeState* state,
Review Comment:
warning: function 'revoke_memory' exceeds recommended size/complexity
thresholds [readability-function-size]
```cpp
Status SpillSortSinkLocalState::revoke_memory(RuntimeState* state,
^
```
<details>
<summary>Additional context</summary>
**be/src/pipeline/exec/spill_sort_sink_operator.cpp:189:** 109 lines
including whitespace and comments (threshold 80)
```cpp
Status SpillSortSinkLocalState::revoke_memory(RuntimeState* state,
^
```
</details>
##########
be/src/pipeline/exec/partitioned_aggregation_source_operator.cpp:
##########
@@ -190,109 +235,112 @@
return source_local_state->open(state);
}
-Status
PartitionedAggLocalState::initiate_merge_spill_partition_agg_data(RuntimeState*
state) {
- DCHECK(!_is_merging);
-
Base::_shared_state->in_mem_shared_state->aggregate_data_container->init_once();
- if
(Base::_shared_state->in_mem_shared_state->aggregate_data_container->iterator !=
-
Base::_shared_state->in_mem_shared_state->aggregate_data_container->end() ||
- _shared_state->spill_partitions.empty()) {
+Status PartitionedAggLocalState::recover_blocks_from_disk(RuntimeState* state,
bool& has_data) {
Review Comment:
warning: function 'recover_blocks_from_disk' exceeds recommended
size/complexity thresholds [readability-function-size]
```cpp
Status PartitionedAggLocalState::recover_blocks_from_disk(RuntimeState*
state, bool& has_data) {
^
```
<details>
<summary>Additional context</summary>
**be/src/pipeline/exec/partitioned_aggregation_source_operator.cpp:237:**
107 lines including whitespace and comments (threshold 80)
```cpp
Status PartitionedAggLocalState::recover_blocks_from_disk(RuntimeState*
state, bool& has_data) {
^
```
</details>
##########
be/src/pipeline/exec/partitioned_hash_join_sink_operator.cpp:
##########
@@ -479,12 +520,9 @@ Status
PartitionedHashJoinSinkOperatorX::_setup_internal_operator(RuntimeState*
Status PartitionedHashJoinSinkOperatorX::sink(RuntimeState* state,
vectorized::Block* in_block,
Review Comment:
warning: function 'sink' has cognitive complexity of 75 (threshold 50)
[readability-function-cognitive-complexity]
```cpp
Status PartitionedHashJoinSinkOperatorX::sink(RuntimeState* state,
vectorized::Block* in_block,
^
```
<details>
<summary>Additional context</summary>
**be/src/pipeline/exec/partitioned_hash_join_sink_operator.cpp:531:** +1,
including nesting penalty of 0, nesting level increased to 1
```cpp
if (rows == 0) {
^
```
**be/src/pipeline/exec/partitioned_hash_join_sink_operator.cpp:532:** +2,
including nesting penalty of 1, nesting level increased to 2
```cpp
if (eos) {
^
```
**be/src/pipeline/exec/partitioned_hash_join_sink_operator.cpp:537:** +3,
including nesting penalty of 2, nesting level increased to 3
```cpp
if (need_to_spill) {
^
```
**be/src/pipeline/exec/partitioned_hash_join_sink_operator.cpp:539:** +1,
nesting level increased to 3
```cpp
} else {
^
```
**be/src/pipeline/exec/partitioned_hash_join_sink_operator.cpp:540:** +4,
including nesting penalty of 3, nesting level increased to 4
```cpp
if
(UNLIKELY(!local_state._shared_state->inner_runtime_state)) {
^
```
**be/src/pipeline/exec/partitioned_hash_join_sink_operator.cpp:541:** +5,
including nesting penalty of 4, nesting level increased to 5
```cpp
RETURN_IF_ERROR(_setup_internal_operator(state));
^
```
**be/src/common/status.h:642:** expanded from macro 'RETURN_IF_ERROR'
```cpp
do { \
^
```
**be/src/pipeline/exec/partitioned_hash_join_sink_operator.cpp:541:** +6,
including nesting penalty of 5, nesting level increased to 6
```cpp
RETURN_IF_ERROR(_setup_internal_operator(state));
^
```
**be/src/common/status.h:644:** expanded from macro 'RETURN_IF_ERROR'
```cpp
if (UNLIKELY(!_status_.ok())) { \
^
```
**be/src/pipeline/exec/partitioned_hash_join_sink_operator.cpp:543:** +4,
including nesting penalty of 3, nesting level increased to 4
```cpp
DBUG_EXECUTE_IF("fault_inject::partitioned_hash_join_sink::sink_eos", {
^
```
**be/src/util/debug_points.h:36:** expanded from macro 'DBUG_EXECUTE_IF'
```cpp
if (UNLIKELY(config::enable_debug_points)) {
\
^
```
**be/src/pipeline/exec/partitioned_hash_join_sink_operator.cpp:543:** +5,
including nesting penalty of 4, nesting level increased to 5
```cpp
DBUG_EXECUTE_IF("fault_inject::partitioned_hash_join_sink::sink_eos", {
^
```
**be/src/util/debug_points.h:38:** expanded from macro 'DBUG_EXECUTE_IF'
```cpp
if (dp) {
\
^
```
**be/src/pipeline/exec/partitioned_hash_join_sink_operator.cpp:549:** +4,
including nesting penalty of 3, nesting level increased to 4
```cpp
RETURN_IF_ERROR(_inner_sink_operator->sink(
^
```
**be/src/common/status.h:642:** expanded from macro 'RETURN_IF_ERROR'
```cpp
do { \
^
```
**be/src/pipeline/exec/partitioned_hash_join_sink_operator.cpp:549:** +5,
including nesting penalty of 4, nesting level increased to 5
```cpp
RETURN_IF_ERROR(_inner_sink_operator->sink(
^
```
**be/src/common/status.h:644:** expanded from macro 'RETURN_IF_ERROR'
```cpp
if (UNLIKELY(!_status_.ok())) { \
^
```
**be/src/pipeline/exec/partitioned_hash_join_sink_operator.cpp:560:**
nesting level increased to 3
```cpp
[&](auto& block) {
^
```
**be/src/pipeline/exec/partitioned_hash_join_sink_operator.cpp:561:** +4,
including nesting penalty of 3, nesting level increased to 4
```cpp
if (block) {
^
```
**be/src/pipeline/exec/partitioned_hash_join_sink_operator.cpp:571:** +1,
including nesting penalty of 0, nesting level increased to 1
```cpp
if (need_to_spill) {
^
```
**be/src/pipeline/exec/partitioned_hash_join_sink_operator.cpp:572:** +2,
including nesting penalty of 1, nesting level increased to 2
```cpp
RETURN_IF_ERROR(local_state._partition_block(state, in_block, 0,
rows));
^
```
**be/src/common/status.h:642:** expanded from macro 'RETURN_IF_ERROR'
```cpp
do { \
^
```
**be/src/pipeline/exec/partitioned_hash_join_sink_operator.cpp:572:** +3,
including nesting penalty of 2, nesting level increased to 3
```cpp
RETURN_IF_ERROR(local_state._partition_block(state, in_block, 0,
rows));
^
```
**be/src/common/status.h:644:** expanded from macro 'RETURN_IF_ERROR'
```cpp
if (UNLIKELY(!_status_.ok())) { \
^
```
**be/src/pipeline/exec/partitioned_hash_join_sink_operator.cpp:573:** +2,
including nesting penalty of 1, nesting level increased to 2
```cpp
if (eos) {
^
```
**be/src/pipeline/exec/partitioned_hash_join_sink_operator.cpp:575:** +1,
nesting level increased to 2
```cpp
} else if (revocable_mem_size(state) >
vectorized::SpillStream::MAX_SPILL_WRITE_BATCH_MEM) {
^
```
**be/src/pipeline/exec/partitioned_hash_join_sink_operator.cpp:578:** +1,
nesting level increased to 1
```cpp
} else {
^
```
**be/src/pipeline/exec/partitioned_hash_join_sink_operator.cpp:579:** +2,
including nesting penalty of 1, nesting level increased to 2
```cpp
if (UNLIKELY(!local_state._shared_state->inner_runtime_state)) {
^
```
**be/src/pipeline/exec/partitioned_hash_join_sink_operator.cpp:580:** +3,
including nesting penalty of 2, nesting level increased to 3
```cpp
RETURN_IF_ERROR(_setup_internal_operator(state));
^
```
**be/src/common/status.h:642:** expanded from macro 'RETURN_IF_ERROR'
```cpp
do { \
^
```
**be/src/pipeline/exec/partitioned_hash_join_sink_operator.cpp:580:** +4,
including nesting penalty of 3, nesting level increased to 4
```cpp
RETURN_IF_ERROR(_setup_internal_operator(state));
^
```
**be/src/common/status.h:644:** expanded from macro 'RETURN_IF_ERROR'
```cpp
if (UNLIKELY(!_status_.ok())) { \
^
```
**be/src/pipeline/exec/partitioned_hash_join_sink_operator.cpp:582:** +2,
including nesting penalty of 1, nesting level increased to 2
```cpp
DBUG_EXECUTE_IF("fault_inject::partitioned_hash_join_sink::sink", {
^
```
**be/src/util/debug_points.h:36:** expanded from macro 'DBUG_EXECUTE_IF'
```cpp
if (UNLIKELY(config::enable_debug_points)) {
\
^
```
**be/src/pipeline/exec/partitioned_hash_join_sink_operator.cpp:582:** +3,
including nesting penalty of 2, nesting level increased to 3
```cpp
DBUG_EXECUTE_IF("fault_inject::partitioned_hash_join_sink::sink", {
^
```
**be/src/util/debug_points.h:38:** expanded from macro 'DBUG_EXECUTE_IF'
```cpp
if (dp) {
\
^
```
**be/src/pipeline/exec/partitioned_hash_join_sink_operator.cpp:587:** +2,
including nesting penalty of 1, nesting level increased to 2
```cpp
RETURN_IF_ERROR(_inner_sink_operator->sink(
^
```
**be/src/common/status.h:642:** expanded from macro 'RETURN_IF_ERROR'
```cpp
do { \
^
```
**be/src/pipeline/exec/partitioned_hash_join_sink_operator.cpp:587:** +3,
including nesting penalty of 2, nesting level increased to 3
```cpp
RETURN_IF_ERROR(_inner_sink_operator->sink(
^
```
**be/src/common/status.h:644:** expanded from macro 'RETURN_IF_ERROR'
```cpp
if (UNLIKELY(!_status_.ok())) { \
^
```
**be/src/pipeline/exec/partitioned_hash_join_sink_operator.cpp:590:** +2,
including nesting penalty of 1, nesting level increased to 2
```cpp
if (eos) {
^
```
</details>
##########
be/src/pipeline/exec/spill_sort_sink_operator.cpp:
##########
@@ -164,88 +168,79 @@ Status SpillSortSinkOperatorX::sink(doris::RuntimeState*
state, vectorized::Bloc
if (eos) {
if (local_state._shared_state->is_spilled) {
if (revocable_mem_size(state) > 0) {
- RETURN_IF_ERROR(revoke_memory(state));
+ RETURN_IF_ERROR(revoke_memory(state, nullptr));
} else {
local_state._dependency->set_ready_to_read();
- local_state._finish_dependency->set_ready();
}
} else {
RETURN_IF_ERROR(
local_state._shared_state->in_mem_shared_state->sorter->prepare_for_read());
local_state._dependency->set_ready_to_read();
- local_state._finish_dependency->set_ready();
}
}
return Status::OK();
}
-Status SpillSortSinkLocalState::revoke_memory(RuntimeState* state) {
+size_t SpillSortSinkLocalState::get_reserve_mem_size(RuntimeState* state, bool
eos) {
+ auto& parent = Base::_parent->template cast<Parent>();
+ return
parent._sort_sink_operator->get_reserve_mem_size_for_next_sink(_runtime_state.get(),
+ eos);
+}
+
+Status SpillSortSinkLocalState::revoke_memory(RuntimeState* state,
Review Comment:
warning: function 'revoke_memory' has cognitive complexity of 51 (threshold
50) [readability-function-cognitive-complexity]
```cpp
Status SpillSortSinkLocalState::revoke_memory(RuntimeState* state,
^
```
<details>
<summary>Additional context</summary>
**be/src/pipeline/exec/spill_sort_sink_operator.cpp:191:** +1, including
nesting penalty of 0, nesting level increased to 1
```cpp
if (!_shared_state->is_spilled) {
^
```
**be/src/pipeline/exec/spill_sort_sink_operator.cpp:204:** +1, including
nesting penalty of 0, nesting level increased to 1
```cpp
RETURN_IF_ERROR(status);
^
```
**be/src/common/status.h:642:** expanded from macro 'RETURN_IF_ERROR'
```cpp
do { \
^
```
**be/src/pipeline/exec/spill_sort_sink_operator.cpp:204:** +2, including
nesting penalty of 1, nesting level increased to 2
```cpp
RETURN_IF_ERROR(status);
^
```
**be/src/common/status.h:644:** expanded from macro 'RETURN_IF_ERROR'
```cpp
if (UNLIKELY(!_status_.ok())) { \
^
```
**be/src/pipeline/exec/spill_sort_sink_operator.cpp:211:** +1, including
nesting penalty of 0, nesting level increased to 1
```cpp
if (!_eos) {
^
```
**be/src/pipeline/exec/spill_sort_sink_operator.cpp:216:** nesting level
increased to 1
```cpp
auto spill_func = [this, state, query_id, &parent] {
^
```
**be/src/pipeline/exec/spill_sort_sink_operator.cpp:242:** +2, including
nesting penalty of 1, nesting level increased to 2
```cpp
RETURN_IF_ERROR(status);
^
```
**be/src/common/status.h:642:** expanded from macro 'RETURN_IF_ERROR'
```cpp
do { \
^
```
**be/src/pipeline/exec/spill_sort_sink_operator.cpp:242:** +3, including
nesting penalty of 2, nesting level increased to 3
```cpp
RETURN_IF_ERROR(status);
^
```
**be/src/common/status.h:644:** expanded from macro 'RETURN_IF_ERROR'
```cpp
if (UNLIKELY(!_status_.ok())) { \
^
```
**be/src/pipeline/exec/spill_sort_sink_operator.cpp:249:** +2, including
nesting penalty of 1, nesting level increased to 2
```cpp
while (!eos && !state->is_cancelled()) {
^
```
**be/src/pipeline/exec/spill_sort_sink_operator.cpp:249:** +1
```cpp
while (!eos && !state->is_cancelled()) {
^
```
**be/src/pipeline/exec/spill_sort_sink_operator.cpp:256:** +3, including
nesting penalty of 2, nesting level increased to 3
```cpp
RETURN_IF_ERROR(status);
^
```
**be/src/common/status.h:642:** expanded from macro 'RETURN_IF_ERROR'
```cpp
do { \
^
```
**be/src/pipeline/exec/spill_sort_sink_operator.cpp:256:** +4, including
nesting penalty of 3, nesting level increased to 4
```cpp
RETURN_IF_ERROR(status);
^
```
**be/src/common/status.h:644:** expanded from macro 'RETURN_IF_ERROR'
```cpp
if (UNLIKELY(!_status_.ok())) { \
^
```
**be/src/pipeline/exec/spill_sort_sink_operator.cpp:258:** +3, including
nesting penalty of 2, nesting level increased to 3
```cpp
RETURN_IF_ERROR(status);
^
```
**be/src/common/status.h:642:** expanded from macro 'RETURN_IF_ERROR'
```cpp
do { \
^
```
**be/src/pipeline/exec/spill_sort_sink_operator.cpp:258:** +4, including
nesting penalty of 3, nesting level increased to 4
```cpp
RETURN_IF_ERROR(status);
^
```
**be/src/common/status.h:644:** expanded from macro 'RETURN_IF_ERROR'
```cpp
if (UNLIKELY(!_status_.ok())) { \
^
```
**be/src/pipeline/exec/spill_sort_sink_operator.cpp:266:** nesting level
increased to 1
```cpp
auto exception_catch_func = [query_id, spill_func]() {
^
```
**be/src/pipeline/exec/spill_sort_sink_operator.cpp:267:** +2, including
nesting penalty of 1, nesting level increased to 2
```cpp
DBUG_EXECUTE_IF("fault_inject::spill_sort_sink::revoke_memory_cancel", {
^
```
**be/src/util/debug_points.h:36:** expanded from macro 'DBUG_EXECUTE_IF'
```cpp
if (UNLIKELY(config::enable_debug_points)) {
\
^
```
**be/src/pipeline/exec/spill_sort_sink_operator.cpp:267:** +3, including
nesting penalty of 2, nesting level increased to 3
```cpp
DBUG_EXECUTE_IF("fault_inject::spill_sort_sink::revoke_memory_cancel", {
^
```
**be/src/util/debug_points.h:38:** expanded from macro 'DBUG_EXECUTE_IF'
```cpp
if (dp) {
\
^
```
**be/src/pipeline/exec/spill_sort_sink_operator.cpp:275:** nesting level
increased to 2
```cpp
auto status = [&]() { RETURN_IF_CATCH_EXCEPTION({ return
spill_func(); }); }();
^
```
**be/src/pipeline/exec/spill_sort_sink_operator.cpp:275:** +3, including
nesting penalty of 2, nesting level increased to 3
```cpp
auto status = [&]() { RETURN_IF_CATCH_EXCEPTION({ return
spill_func(); }); }();
^
```
**be/src/common/exception.h:79:** expanded from macro
'RETURN_IF_CATCH_EXCEPTION'
```cpp
do {
\
^
```
**be/src/pipeline/exec/spill_sort_sink_operator.cpp:275:** +4, including
nesting penalty of 3, nesting level increased to 4
```cpp
auto status = [&]() { RETURN_IF_CATCH_EXCEPTION({ return
spill_func(); }); }();
^
```
**be/src/common/exception.h:84:** expanded from macro
'RETURN_IF_CATCH_EXCEPTION'
```cpp
} catch (const doris::Exception& e) {
\
^
```
**be/src/pipeline/exec/spill_sort_sink_operator.cpp:275:** +5, including
nesting penalty of 4, nesting level increased to 5
```cpp
auto status = [&]() { RETURN_IF_CATCH_EXCEPTION({ return
spill_func(); }); }();
^
```
**be/src/common/exception.h:85:** expanded from macro
'RETURN_IF_CATCH_EXCEPTION'
```cpp
if (e.code() == doris::ErrorCode::MEM_ALLOC_FAILED) {
\
^
```
**be/src/pipeline/exec/spill_sort_sink_operator.cpp:280:** +1, including
nesting penalty of 0, nesting level increased to 1
```cpp
DBUG_EXECUTE_IF("fault_inject::spill_sort_sink::revoke_memory_submit_func", {
^
```
**be/src/util/debug_points.h:36:** expanded from macro 'DBUG_EXECUTE_IF'
```cpp
if (UNLIKELY(config::enable_debug_points)) {
\
^
```
**be/src/pipeline/exec/spill_sort_sink_operator.cpp:280:** +2, including
nesting penalty of 1, nesting level increased to 2
```cpp
DBUG_EXECUTE_IF("fault_inject::spill_sort_sink::revoke_memory_submit_func", {
^
```
**be/src/util/debug_points.h:38:** expanded from macro 'DBUG_EXECUTE_IF'
```cpp
if (dp) {
\
^
```
**be/src/pipeline/exec/spill_sort_sink_operator.cpp:285:** +1, including
nesting penalty of 0, nesting level increased to 1
```cpp
if (status.ok()) {
^
```
**be/src/pipeline/exec/spill_sort_sink_operator.cpp:293:** +1, including
nesting penalty of 0, nesting level increased to 1
```cpp
if (!status.ok()) {
^
```
**be/src/pipeline/exec/spill_sort_sink_operator.cpp:294:** +2, including
nesting penalty of 1, nesting level increased to 2
```cpp
if (!_eos) {
^
```
</details>
##########
be/src/pipeline/exec/partitioned_hash_join_probe_operator.cpp:
##########
@@ -654,32 +682,50 @@ Status
PartitionedHashJoinProbeOperatorX::_setup_internal_operators(
return Status::Error<INTERNAL_ERROR>(
"fault_inject partitioned_hash_join_probe sink failed");
});
-
RETURN_IF_ERROR(_inner_sink_operator->sink(local_state._runtime_state.get(),
&block, true));
- VLOG_DEBUG << "query: " << print_id(state->query_id())
+
+
RETURN_IF_ERROR(_inner_sink_operator->sink(local_state._shared_state->inner_runtime_state.get(),
+ &block, true));
+ VLOG_DEBUG << "Query: " << print_id(state->query_id())
<< ", internal build operator finished, node id: " << node_id()
<< ", task id: " << state->task_id()
- << ", partition: " << local_state._partition_cursor;
+ << ", partition: " << local_state._partition_cursor << "rows: "
<< block.rows()
+ << ", usage: "
+ << _inner_sink_operator->get_memory_usage(
+
local_state._shared_state->inner_runtime_state.get());
+
+ COUNTER_SET(local_state._hash_table_memory_usage,
+
sink_local_state->profile()->get_counter("MemoryUsageHashTable")->value());
return Status::OK();
}
Status PartitionedHashJoinProbeOperatorX::pull(doris::RuntimeState* state,
Review Comment:
warning: function 'pull' has cognitive complexity of 72 (threshold 50)
[readability-function-cognitive-complexity]
```cpp
Status PartitionedHashJoinProbeOperatorX::pull(doris::RuntimeState* state,
^
```
<details>
<summary>Additional context</summary>
**be/src/pipeline/exec/partitioned_hash_join_probe_operator.cpp:707:** +1,
including nesting penalty of 0, nesting level increased to 1
```cpp
if (local_state._recovered_build_block &&
!local_state._recovered_build_block->empty()) {
^
```
**be/src/pipeline/exec/partitioned_hash_join_probe_operator.cpp:707:** +1
```cpp
if (local_state._recovered_build_block &&
!local_state._recovered_build_block->empty()) {
^
```
**be/src/pipeline/exec/partitioned_hash_join_probe_operator.cpp:710:** +2,
including nesting penalty of 1, nesting level increased to 2
```cpp
if (!mutable_block) {
^
```
**be/src/pipeline/exec/partitioned_hash_join_probe_operator.cpp:712:** +1,
nesting level increased to 2
```cpp
} else {
^
```
**be/src/pipeline/exec/partitioned_hash_join_probe_operator.cpp:713:** +3,
including nesting penalty of 2, nesting level increased to 3
```cpp
RETURN_IF_ERROR(mutable_block->merge(local_state._recovered_build_block->to_block()));
^
```
**be/src/common/status.h:642:** expanded from macro 'RETURN_IF_ERROR'
```cpp
do { \
^
```
**be/src/pipeline/exec/partitioned_hash_join_probe_operator.cpp:713:** +4,
including nesting penalty of 3, nesting level increased to 4
```cpp
RETURN_IF_ERROR(mutable_block->merge(local_state._recovered_build_block->to_block()));
^
```
**be/src/common/status.h:644:** expanded from macro 'RETURN_IF_ERROR'
```cpp
if (UNLIKELY(!_status_.ok())) { \
^
```
**be/src/pipeline/exec/partitioned_hash_join_probe_operator.cpp:718:** +1,
including nesting penalty of 0, nesting level increased to 1
```cpp
if (local_state._need_to_setup_internal_operators) {
^
```
**be/src/pipeline/exec/partitioned_hash_join_probe_operator.cpp:720:** +2,
including nesting penalty of 1, nesting level increased to 2
```cpp
RETURN_IF_ERROR(local_state.recover_build_blocks_from_disk(
^
```
**be/src/common/status.h:642:** expanded from macro 'RETURN_IF_ERROR'
```cpp
do { \
^
```
**be/src/pipeline/exec/partitioned_hash_join_probe_operator.cpp:720:** +3,
including nesting penalty of 2, nesting level increased to 3
```cpp
RETURN_IF_ERROR(local_state.recover_build_blocks_from_disk(
^
```
**be/src/common/status.h:644:** expanded from macro 'RETURN_IF_ERROR'
```cpp
if (UNLIKELY(!_status_.ok())) { \
^
```
**be/src/pipeline/exec/partitioned_hash_join_probe_operator.cpp:722:** +2,
including nesting penalty of 1, nesting level increased to 2
```cpp
if (has_data) {
^
```
**be/src/pipeline/exec/partitioned_hash_join_probe_operator.cpp:727:** +2,
including nesting penalty of 1, nesting level increased to 2
```cpp
RETURN_IF_ERROR(local_state.finish_spilling(partition_index));
^
```
**be/src/common/status.h:642:** expanded from macro 'RETURN_IF_ERROR'
```cpp
do { \
^
```
**be/src/pipeline/exec/partitioned_hash_join_probe_operator.cpp:727:** +3,
including nesting penalty of 2, nesting level increased to 3
```cpp
RETURN_IF_ERROR(local_state.finish_spilling(partition_index));
^
```
**be/src/common/status.h:644:** expanded from macro 'RETURN_IF_ERROR'
```cpp
if (UNLIKELY(!_status_.ok())) { \
^
```
**be/src/pipeline/exec/partitioned_hash_join_probe_operator.cpp:728:** +2,
including nesting penalty of 1, nesting level increased to 2
```cpp
RETURN_IF_ERROR(_setup_internal_operators(local_state, state));
^
```
**be/src/common/status.h:642:** expanded from macro 'RETURN_IF_ERROR'
```cpp
do { \
^
```
**be/src/pipeline/exec/partitioned_hash_join_probe_operator.cpp:728:** +3,
including nesting penalty of 2, nesting level increased to 3
```cpp
RETURN_IF_ERROR(_setup_internal_operators(local_state, state));
^
```
**be/src/common/status.h:644:** expanded from macro 'RETURN_IF_ERROR'
```cpp
if (UNLIKELY(!_status_.ok())) { \
^
```
**be/src/pipeline/exec/partitioned_hash_join_probe_operator.cpp:731:** +2,
including nesting penalty of 1, nesting level increased to 2
```cpp
if (mutable_block && !mutable_block->empty()) {
^
```
**be/src/pipeline/exec/partitioned_hash_join_probe_operator.cpp:731:** +1
```cpp
if (mutable_block && !mutable_block->empty()) {
^
```
**be/src/pipeline/exec/partitioned_hash_join_probe_operator.cpp:737:** +1,
including nesting penalty of 0, nesting level increased to 1
```cpp
while (_inner_probe_operator->need_more_input_data(runtime_state)) {
^
```
**be/src/pipeline/exec/partitioned_hash_join_probe_operator.cpp:738:** +2,
including nesting penalty of 1, nesting level increased to 2
```cpp
if (probe_blocks.empty()) {
^
```
**be/src/pipeline/exec/partitioned_hash_join_probe_operator.cpp:741:** +3,
including nesting penalty of 2, nesting level increased to 3
```cpp
RETURN_IF_ERROR(
^
```
**be/src/common/status.h:642:** expanded from macro 'RETURN_IF_ERROR'
```cpp
do { \
^
```
**be/src/pipeline/exec/partitioned_hash_join_probe_operator.cpp:741:** +4,
including nesting penalty of 3, nesting level increased to 4
```cpp
RETURN_IF_ERROR(
^
```
**be/src/common/status.h:644:** expanded from macro 'RETURN_IF_ERROR'
```cpp
if (UNLIKELY(!_status_.ok())) { \
^
```
**be/src/pipeline/exec/partitioned_hash_join_probe_operator.cpp:743:** +3,
including nesting penalty of 2, nesting level increased to 3
```cpp
if (!has_data) {
^
```
**be/src/pipeline/exec/partitioned_hash_join_probe_operator.cpp:745:** +4,
including nesting penalty of 3, nesting level increased to 4
```cpp
RETURN_IF_ERROR(_inner_probe_operator->push(runtime_state,
&block, true));
^
```
**be/src/common/status.h:642:** expanded from macro 'RETURN_IF_ERROR'
```cpp
do { \
^
```
**be/src/pipeline/exec/partitioned_hash_join_probe_operator.cpp:745:** +5,
including nesting penalty of 4, nesting level increased to 5
```cpp
RETURN_IF_ERROR(_inner_probe_operator->push(runtime_state,
&block, true));
^
```
**be/src/common/status.h:644:** expanded from macro 'RETURN_IF_ERROR'
```cpp
if (UNLIKELY(!_status_.ok())) { \
^
```
**be/src/pipeline/exec/partitioned_hash_join_probe_operator.cpp:750:** +1,
nesting level increased to 3
```cpp
} else {
^
```
**be/src/pipeline/exec/partitioned_hash_join_probe_operator.cpp:757:** +2,
including nesting penalty of 1, nesting level increased to 2
```cpp
if (!block.empty()) {
^
```
**be/src/pipeline/exec/partitioned_hash_join_probe_operator.cpp:758:** +3,
including nesting penalty of 2, nesting level increased to 3
```cpp
RETURN_IF_ERROR(_inner_probe_operator->push(runtime_state,
&block, false));
^
```
**be/src/common/status.h:642:** expanded from macro 'RETURN_IF_ERROR'
```cpp
do { \
^
```
**be/src/pipeline/exec/partitioned_hash_join_probe_operator.cpp:758:** +4,
including nesting penalty of 3, nesting level increased to 4
```cpp
RETURN_IF_ERROR(_inner_probe_operator->push(runtime_state,
&block, false));
^
```
**be/src/common/status.h:644:** expanded from macro 'RETURN_IF_ERROR'
```cpp
if (UNLIKELY(!_status_.ok())) { \
^
```
**be/src/pipeline/exec/partitioned_hash_join_probe_operator.cpp:762:** +1,
including nesting penalty of 0, nesting level increased to 1
```cpp
RETURN_IF_ERROR(_inner_probe_operator->pull(
^
```
**be/src/common/status.h:642:** expanded from macro 'RETURN_IF_ERROR'
```cpp
do { \
^
```
**be/src/pipeline/exec/partitioned_hash_join_probe_operator.cpp:762:** +2,
including nesting penalty of 1, nesting level increased to 2
```cpp
RETURN_IF_ERROR(_inner_probe_operator->pull(
^
```
**be/src/common/status.h:644:** expanded from macro 'RETURN_IF_ERROR'
```cpp
if (UNLIKELY(!_status_.ok())) { \
^
```
**be/src/pipeline/exec/partitioned_hash_join_probe_operator.cpp:766:** +1,
including nesting penalty of 0, nesting level increased to 1
```cpp
if (in_mem_eos) {
^
```
**be/src/pipeline/exec/partitioned_hash_join_probe_operator.cpp:771:** +2,
including nesting penalty of 1, nesting level increased to 2
```cpp
if (local_state._partition_cursor == _partition_count) {
^
```
**be/src/pipeline/exec/partitioned_hash_join_probe_operator.cpp:773:** +1,
nesting level increased to 2
```cpp
} else {
^
```
</details>
##########
be/src/runtime/workload_group/workload_group_manager.cpp:
##########
@@ -297,7 +266,590 @@ void
WorkloadGroupMgr::get_wg_resource_usage(vectorized::Block* block) {
SchemaScannerHelper::insert_int64_value(4,
wg->get_local_scan_bytes_per_second(), block);
SchemaScannerHelper::insert_int64_value(5,
wg->get_remote_scan_bytes_per_second(), block);
+ SchemaScannerHelper::insert_int64_value(6, wg->write_buffer_size(),
block);
+ }
+}
+
+void WorkloadGroupMgr::add_paused_query(const std::shared_ptr<QueryContext>&
query_ctx,
+ int64_t reserve_size) {
+ std::lock_guard<std::mutex> lock(_paused_queries_lock);
+ DCHECK(query_ctx != nullptr);
+ auto wg = query_ctx->workload_group();
+ auto&& [it, inserted] = _paused_queries_list[wg].emplace(
+ query_ctx,
doris::GlobalMemoryArbitrator::last_affected_cache_capacity_adjust_weighted,
+ doris::GlobalMemoryArbitrator::any_workload_group_exceed_limit,
reserve_size);
+ // Check if this is an invalid reserve, for example, if the reserve size
is too large, larger than the query limit
+ // if hard limit is enabled, then not need enable other queries hard limit.
+ if (inserted) {
+ query_ctx->set_memory_sufficient(false);
+ LOG(INFO) << "Insert one new paused query: " <<
query_ctx->debug_string()
+ << ", workload group: " << wg->debug_string();
+ }
+}
+
+/**
+ * Strategy 1: A revocable query should not have any running
task(PipelineTask).
+ * strategy 2: If the workload group has any task exceed workload group
memlimit, then set all queryctx's memlimit
+ * strategy 3: If any query exceed process memlimit, then should clear all
caches.
+ * strategy 4: If any query exceed query's memlimit, then do spill disk or
cancel it.
+ * strategy 5: If any query exceed process's memlimit and cache is zero, then
do following:
+ */
+void WorkloadGroupMgr::handle_paused_queries() {
+ {
+ std::shared_lock<std::shared_mutex> r_lock(_group_mutex);
+ for (auto& [wg_id, wg] : _workload_groups) {
+ std::unique_lock<std::mutex> lock(_paused_queries_lock);
+ if (_paused_queries_list[wg].empty()) {
+ // Add an empty set to wg that not contains paused queries.
+ }
+ }
+ }
+
+ std::unique_lock<std::mutex> lock(_paused_queries_lock);
+ bool has_revoked_from_other_group = false;
+ for (auto it = _paused_queries_list.begin(); it !=
_paused_queries_list.end();) {
+ auto& queries_list = it->second;
+ const auto& wg = it->first;
+
+ LOG_EVERY_T(INFO, 120) << "Paused queries count: " <<
queries_list.size();
+
+ bool is_low_watermark = false;
+ bool is_high_watermark = false;
+ wg->check_mem_used(&is_low_watermark, &is_high_watermark);
+
+ bool has_changed_hard_limit = false;
+ int64_t flushed_memtable_bytes = 0;
+ // If the query is paused because its limit exceed the query itself's
memlimit, then just spill disk.
+ // The query's memlimit is set using slot mechanism and its value is
set using the user settings, not
+ // by weighted value. So if reserve failed, then it is actually exceed
limit.
+ for (auto query_it = queries_list.begin(); query_it !=
queries_list.end();) {
+ auto query_ctx = query_it->query_ctx_.lock();
+ // The query is finished during in paused list.
+ if (query_ctx == nullptr) {
+ LOG(INFO) << "Query: " << query_it->query_id() << " is
nullptr, erase it.";
+ query_it = queries_list.erase(query_it);
+ continue;
+ }
+ if (query_ctx->is_cancelled()) {
+ LOG(INFO) << "Query: " << print_id(query_ctx->query_id())
+ << " was canceled, remove from paused list";
+ query_it = queries_list.erase(query_it);
+ continue;
+ }
+
+ if
(query_ctx->paused_reason().is<ErrorCode::QUERY_MEMORY_EXCEEDED>()) {
+ // Streamload, kafka load, group commit will never have query
memory exceeded error because
+ // their query limit is very large.
+ bool spill_res =
+ handle_single_query_(query_ctx,
query_it->reserve_size_,
+ query_it->elapsed_time(),
query_ctx->paused_reason());
+ if (!spill_res) {
+ ++query_it;
+ continue;
+ } else {
+ query_it = queries_list.erase(query_it);
+ continue;
+ }
+ } else if
(query_ctx->paused_reason().is<ErrorCode::WORKLOAD_GROUP_MEMORY_EXCEEDED>()) {
+ // Only deal with non overcommit workload group.
+ if (wg->enable_memory_overcommit()) {
+ // Soft limit wg will only reserve failed when process
limit exceed. But in some corner case,
+ // when reserve, the wg is hard limit, the query reserve
failed, but when this loop run
+ // the wg is converted to soft limit.
+ // So that should resume the query.
+ LOG(WARNING)
+ << "Query: " << print_id(query_ctx->query_id())
+ << " reserve memory failed because exceed workload
group memlimit, it "
+ "should not happen, resume it again. paused
reason: "
+ << query_ctx->paused_reason();
+ query_ctx->set_memory_sufficient(true);
+ query_it = queries_list.erase(query_it);
+ continue;
+ }
+ // check if the reserve is too large, if it is too large,
+ // should set the query's limit only.
+ // Check the query's reserve with expected limit.
+ if (query_ctx->expected_mem_limit() <
+ query_ctx->get_mem_tracker()->consumption() +
query_it->reserve_size_) {
+ query_ctx->set_mem_limit(query_ctx->expected_mem_limit());
+ query_ctx->set_memory_sufficient(true);
+ LOG(INFO) << "Workload group memory reserve failed because
"
+ << query_ctx->debug_string() << " reserve size "
+ <<
PrettyPrinter::print_bytes(query_it->reserve_size_)
+ << " is too large, set hard limit to "
+ <<
PrettyPrinter::print_bytes(query_ctx->expected_mem_limit())
+ << " and resume running.";
+ query_it = queries_list.erase(query_it);
+ continue;
+ }
+ if (flushed_memtable_bytes <= 0) {
+ flushed_memtable_bytes =
+ flush_memtable_from_current_group_(wg,
query_it->reserve_size_);
+ }
+ if (flushed_memtable_bytes > 0) {
+ // Flushed some memtable, just wait flush finished and not
do anything more.
+ wg->enable_write_buffer_limit(true);
+ ++query_it;
+ continue;
+ }
+ if (!has_changed_hard_limit) {
+ update_queries_limit_(wg, true);
+ has_changed_hard_limit = true;
+ LOG(INFO) << "Query: " << print_id(query_ctx->query_id())
<< " reserve memory("
+ <<
PrettyPrinter::print_bytes(query_it->reserve_size_)
+ << ") failed due to workload group memory
exceed, "
+ "should set the workload group work in memory
insufficent mode, "
+ "so that other query will reduce their
memory. wg: "
+ << wg->debug_string();
+ }
+ if (wg->slot_memory_policy() == TWgSlotMemoryPolicy::NONE) {
+ // If not enable slot memory policy, then should spill
directly
+ // Maybe there are another query that use too much memory,
but we
+ // not encourage not enable slot memory.
+ // TODO should kill the query that exceed limit.
+ bool spill_res = handle_single_query_(query_ctx,
query_it->reserve_size_,
+
query_it->elapsed_time(),
+
query_ctx->paused_reason());
+ if (!spill_res) {
+ ++query_it;
+ continue;
+ } else {
+ query_it = queries_list.erase(query_it);
+ continue;
+ }
+ } else {
+ // Should not put the query back to task scheduler
immediately, because when wg's memory not sufficient,
+ // and then set wg's flag, other query may not free memory
very quickly.
+ if (query_it->elapsed_time() >
config::spill_in_paused_queue_timeout_ms) {
+ // set wg's memory to insufficent, then add it back to
task scheduler to run.
+ LOG(INFO) << "Query: " <<
print_id(query_ctx->query_id())
+ << " will be resume.";
+ query_ctx->set_memory_sufficient(true);
+ query_it = queries_list.erase(query_it);
+ continue;
+ } else {
+ ++query_it;
+ continue;
+ }
+ }
+ } else {
+ // If wg's memlimit not exceed, but process memory exceed, it
means cache or other metadata
+ // used too much memory. Should clean all cache here.
+ // 1. Check cache used, if cache is larger than > 0, then just
return and wait for it to 0 to release some memory.
+ if
(doris::GlobalMemoryArbitrator::last_affected_cache_capacity_adjust_weighted >
+ 0.001 &&
+
doris::GlobalMemoryArbitrator::last_wg_trigger_cache_capacity_adjust_weighted >
+ 0.001) {
+
doris::GlobalMemoryArbitrator::last_wg_trigger_cache_capacity_adjust_weighted =
+ 0;
+
doris::GlobalMemoryArbitrator::notify_cache_adjust_capacity();
+ LOG(INFO) << "There are some queries need process memory,
so that set cache "
+ "capacity "
+ "to 0 now";
+ }
+ if (query_it->cache_ratio_ < 0.001) {
+ // 1. Check if could revoke some memory from memtable
+ if (flushed_memtable_bytes <= 0) {
+ flushed_memtable_bytes =
+ flush_memtable_from_current_group_(wg,
query_it->reserve_size_);
+ }
+ if (flushed_memtable_bytes > 0) {
+ // Flushed some memtable, just wait flush finished and
not do anything more.
+ ++query_it;
+ continue;
+ }
+ // TODO should wait here to check if the process has
release revoked_size memory and then continue.
+ if (!has_revoked_from_other_group) {
+ int64_t revoked_size = revoke_memory_from_other_group_(
+ query_ctx, wg->enable_memory_overcommit(),
query_it->reserve_size_);
+ if (revoked_size > 0) {
+ has_revoked_from_other_group = true;
+ query_ctx->set_memory_sufficient(true);
+ query_it = queries_list.erase(query_it);
+ // Do not care if the revoked_size > reserve size,
and try to run again.
+ continue;
+ } else {
+ bool spill_res = handle_single_query_(
+ query_ctx, query_it->reserve_size_,
query_it->elapsed_time(),
+ query_ctx->paused_reason());
+ if (spill_res) {
+ query_it = queries_list.erase(query_it);
+ continue;
+ } else {
+ ++query_it;
+ continue;
+ }
+ }
+ } else {
+ // If any query is cancelled during process limit
stage, should resume other query and
+ // do not do any check now.
+ query_ctx->set_memory_sufficient(true);
+ query_it = queries_list.erase(query_it);
+ continue;
+ }
+ }
+ if
(doris::GlobalMemoryArbitrator::last_affected_cache_capacity_adjust_weighted <
+ 0.001 &&
+ query_it->cache_ratio_ > 0.001) {
+ LOG(INFO) << "Query: " << print_id(query_ctx->query_id())
+ << " will be resume after cache adjust.";
+ query_ctx->set_memory_sufficient(true);
+ query_it = queries_list.erase(query_it);
+ continue;
+ }
+ ++query_it;
+ }
+ }
+ // Not need waiting flush memtable and below low watermark disable
load buffer limit
+ if (flushed_memtable_bytes <= 0 && !is_low_watermark) {
+ wg->enable_write_buffer_limit(false);
+ }
+
+ if (queries_list.empty()) {
+ it = _paused_queries_list.erase(it);
+ continue;
+ } else {
+ // Finished deal with one workload group, and should deal with
next one.
+ ++it;
+ }
+ }
+}
+
+// Return the expected free bytes if memtable could flush
+int64_t WorkloadGroupMgr::flush_memtable_from_current_group_(WorkloadGroupPtr
wg,
+ int64_t
need_free_mem) {
+ // If there are a lot of memtable memory, then wait them flush finished.
+ MemTableMemoryLimiter* memtable_limiter =
+ doris::ExecEnv::GetInstance()->memtable_memory_limiter();
+ int64_t memtable_active_bytes = 0;
+ int64_t memtable_queue_bytes = 0;
+ int64_t memtable_flush_bytes = 0;
+ memtable_limiter->get_workload_group_memtable_usage(
+ wg->id(), &memtable_active_bytes, &memtable_queue_bytes,
&memtable_flush_bytes);
+ // TODO: should add a signal in memtable limiter to prevent new batch
+ // For example, streamload, it will not reserve many memory, but it will
occupy many memtable memory.
+ // TODO: 0.2 should be a workload group properties. For example, the group
is optimized for load,then the value
+ // should be larged, if the group is optimized for query, then the value
should be smaller.
+ int64_t max_wg_memtable_bytes = wg->write_buffer_limit();
+ if (memtable_active_bytes + memtable_queue_bytes + memtable_flush_bytes >
+ max_wg_memtable_bytes) {
+ // There are many table in flush queue, just waiting them flush
finished.
+ if (memtable_active_bytes < (int64_t)(max_wg_memtable_bytes * 0.6)) {
+ LOG_EVERY_T(INFO, 60) << wg->name()
+ << " load memtable size is: " <<
memtable_active_bytes << ", "
+ << memtable_queue_bytes << ", " <<
memtable_flush_bytes
+ << ", load buffer limit is: " <<
max_wg_memtable_bytes
+ << " wait for flush finished to release more
memory";
+ return memtable_queue_bytes + memtable_flush_bytes;
+ } else {
+ // Flush some memtables(currently written) to flush queue.
+ memtable_limiter->flush_workload_group_memtables(
+ wg->id(), memtable_active_bytes -
(int64_t)(max_wg_memtable_bytes * 0.6));
+ LOG_EVERY_T(INFO, 60) << wg->name()
+ << " load memtable size is: " <<
memtable_active_bytes << ", "
+ << memtable_queue_bytes << ", " <<
memtable_flush_bytes
+ << ", flush some active memtable to revoke
memory";
+ return memtable_queue_bytes + memtable_flush_bytes +
memtable_active_bytes -
+ (int64_t)(max_wg_memtable_bytes * 0.6);
+ }
+ }
+ return 0;
+}
+
+int64_t
WorkloadGroupMgr::revoke_memory_from_other_group_(std::shared_ptr<QueryContext>
requestor,
+ bool hard_limit,
int64_t need_free_mem) {
+ int64_t total_freed_mem = 0;
+ std::unique_ptr<RuntimeProfile> profile =
std::make_unique<RuntimeProfile>("RevokeMemory");
+ // 1. memtable like memory
+ // 2. query exceed workload group limit
+ int64_t freed_mem = revoke_overcommited_memory_(requestor, need_free_mem,
profile.get());
+ total_freed_mem += freed_mem;
+ // The revoke process may kill current requestor, so should return now.
+ if (need_free_mem - total_freed_mem < 0 || requestor->is_cancelled()) {
+ return total_freed_mem;
+ }
+ if (hard_limit) {
+ freed_mem = cancel_top_query_in_overcommit_group_(need_free_mem -
total_freed_mem,
+
doris::QUERY_MIN_MEMORY, profile.get());
+ } else {
+ freed_mem = cancel_top_query_in_overcommit_group_(
+ need_free_mem - total_freed_mem,
requestor->get_mem_tracker()->consumption(),
+ profile.get());
+ }
+ total_freed_mem += freed_mem;
+ // The revoke process may kill current requestor, so should return now.
+ if (need_free_mem - total_freed_mem < 0 || requestor->is_cancelled()) {
+ return total_freed_mem;
+ }
+ return total_freed_mem;
+}
+
+// Revoke memory from workload group that exceed it's limit. For example, if
the wg's limit is 10g, but used 12g
+// then should revoke 2g from the group.
+int64_t
WorkloadGroupMgr::revoke_overcommited_memory_(std::shared_ptr<QueryContext>
requestor,
+ int64_t need_free_mem,
+ RuntimeProfile* profile)
{
+ int64_t total_freed_mem = 0;
+ // 1. check memtable usage, and try to free them.
+ int64_t freed_mem =
revoke_memtable_from_overcommited_groups_(need_free_mem, profile);
+ total_freed_mem += freed_mem;
+ // The revoke process may kill current requestor, so should return now.
+ if (need_free_mem - total_freed_mem < 0 || requestor->is_cancelled()) {
+ return total_freed_mem;
+ }
+ // 2. Cancel top usage query, one by one
+ using WorkloadGroupMem = std::pair<WorkloadGroupPtr, int64_t>;
+ auto cmp = [](WorkloadGroupMem left, WorkloadGroupMem right) {
+ return left.second < right.second;
+ };
+ std::priority_queue<WorkloadGroupMem, std::vector<WorkloadGroupMem>,
decltype(cmp)> heap(cmp);
+ {
+ std::shared_lock<std::shared_mutex> r_lock(_group_mutex);
+ for (auto iter = _workload_groups.begin(); iter !=
_workload_groups.end(); iter++) {
+ if (requestor->workload_group() != nullptr &&
+ iter->second->id() == requestor->workload_group()->id()) {
+ continue;
+ }
+ heap.emplace(iter->second, iter->second->memory_used());
+ }
+ }
+ while (!heap.empty() && need_free_mem - total_freed_mem > 0 &&
!requestor->is_cancelled()) {
+ auto [wg, sort_mem] = heap.top();
+ heap.pop();
+ freed_mem = wg->free_overcommited_memory(need_free_mem -
total_freed_mem, profile);
+ total_freed_mem += freed_mem;
+ }
+ return total_freed_mem;
+}
+
+// If the memtable is too large, then flush them and wait for finished.
+int64_t WorkloadGroupMgr::revoke_memtable_from_overcommited_groups_(int64_t
need_free_mem,
+
RuntimeProfile* profile) {
+ return 0;
+}
+
+// 1. Sort all memory limiter in all overcommit wg, and cancel the top usage
task that with most memory.
+// 2. Maybe not valid because it's memory not exceed limit.
+int64_t WorkloadGroupMgr::cancel_top_query_in_overcommit_group_(int64_t
need_free_mem,
+ int64_t
lower_bound,
+
RuntimeProfile* profile) {
+ return 0;
+}
+
+// streamload, kafka routine load, group commit
+// insert into select
+// select
+
+// If the query could release some memory, for example, spill disk, then the
return value is true.
+// If the query could not release memory, then cancel the query, the return
value is true.
+// If the query is not ready to do these tasks, it means just wait, then
return value is false.
+bool WorkloadGroupMgr::handle_single_query_(const
std::shared_ptr<QueryContext>& query_ctx,
Review Comment:
warning: function 'handle_single_query_' exceeds recommended size/complexity
thresholds [readability-function-size]
```cpp
bool WorkloadGroupMgr::handle_single_query_(const
std::shared_ptr<QueryContext>& query_ctx,
^
```
<details>
<summary>Additional context</summary>
**be/src/runtime/workload_group/workload_group_manager.cpp:645:** 105 lines
including whitespace and comments (threshold 80)
```cpp
bool WorkloadGroupMgr::handle_single_query_(const
std::shared_ptr<QueryContext>& query_ctx,
^
```
</details>
##########
be/src/runtime/workload_group/workload_group_manager.cpp:
##########
@@ -297,7 +266,590 @@
SchemaScannerHelper::insert_int64_value(4,
wg->get_local_scan_bytes_per_second(), block);
SchemaScannerHelper::insert_int64_value(5,
wg->get_remote_scan_bytes_per_second(), block);
+ SchemaScannerHelper::insert_int64_value(6, wg->write_buffer_size(),
block);
+ }
+}
+
+void WorkloadGroupMgr::add_paused_query(const std::shared_ptr<QueryContext>&
query_ctx,
+ int64_t reserve_size) {
+ std::lock_guard<std::mutex> lock(_paused_queries_lock);
+ DCHECK(query_ctx != nullptr);
+ auto wg = query_ctx->workload_group();
+ auto&& [it, inserted] = _paused_queries_list[wg].emplace(
+ query_ctx,
doris::GlobalMemoryArbitrator::last_affected_cache_capacity_adjust_weighted,
+ doris::GlobalMemoryArbitrator::any_workload_group_exceed_limit,
reserve_size);
+ // Check if this is an invalid reserve, for example, if the reserve size
is too large, larger than the query limit
+ // if hard limit is enabled, then not need enable other queries hard limit.
+ if (inserted) {
+ query_ctx->set_memory_sufficient(false);
+ LOG(INFO) << "Insert one new paused query: " <<
query_ctx->debug_string()
+ << ", workload group: " << wg->debug_string();
+ }
+}
+
+/**
+ * Strategy 1: A revocable query should not have any running
task(PipelineTask).
+ * strategy 2: If the workload group has any task exceed workload group
memlimit, then set all queryctx's memlimit
+ * strategy 3: If any query exceed process memlimit, then should clear all
caches.
+ * strategy 4: If any query exceed query's memlimit, then do spill disk or
cancel it.
+ * strategy 5: If any query exceed process's memlimit and cache is zero, then
do following:
+ */
+void WorkloadGroupMgr::handle_paused_queries() {
+ {
+ std::shared_lock<std::shared_mutex> r_lock(_group_mutex);
+ for (auto& [wg_id, wg] : _workload_groups) {
+ std::unique_lock<std::mutex> lock(_paused_queries_lock);
+ if (_paused_queries_list[wg].empty()) {
+ // Add an empty set to wg that not contains paused queries.
+ }
+ }
+ }
+
+ std::unique_lock<std::mutex> lock(_paused_queries_lock);
+ bool has_revoked_from_other_group = false;
+ for (auto it = _paused_queries_list.begin(); it !=
_paused_queries_list.end();) {
+ auto& queries_list = it->second;
+ const auto& wg = it->first;
+
+ LOG_EVERY_T(INFO, 120) << "Paused queries count: " <<
queries_list.size();
+
+ bool is_low_watermark = false;
+ bool is_high_watermark = false;
+ wg->check_mem_used(&is_low_watermark, &is_high_watermark);
+
+ bool has_changed_hard_limit = false;
+ int64_t flushed_memtable_bytes = 0;
+ // If the query is paused because its limit exceed the query itself's
memlimit, then just spill disk.
+ // The query's memlimit is set using slot mechanism and its value is
set using the user settings, not
+ // by weighted value. So if reserve failed, then it is actually exceed
limit.
+ for (auto query_it = queries_list.begin(); query_it !=
queries_list.end();) {
+ auto query_ctx = query_it->query_ctx_.lock();
+ // The query is finished during in paused list.
+ if (query_ctx == nullptr) {
+ LOG(INFO) << "Query: " << query_it->query_id() << " is
nullptr, erase it.";
+ query_it = queries_list.erase(query_it);
+ continue;
+ }
+ if (query_ctx->is_cancelled()) {
+ LOG(INFO) << "Query: " << print_id(query_ctx->query_id())
+ << " was canceled, remove from paused list";
+ query_it = queries_list.erase(query_it);
+ continue;
+ }
+
+ if
(query_ctx->paused_reason().is<ErrorCode::QUERY_MEMORY_EXCEEDED>()) {
+ // Streamload, kafka load, group commit will never have query
memory exceeded error because
+ // their query limit is very large.
+ bool spill_res =
+ handle_single_query_(query_ctx,
query_it->reserve_size_,
+ query_it->elapsed_time(),
query_ctx->paused_reason());
+ if (!spill_res) {
+ ++query_it;
+ continue;
+ } else {
+ query_it = queries_list.erase(query_it);
+ continue;
+ }
+ } else if
(query_ctx->paused_reason().is<ErrorCode::WORKLOAD_GROUP_MEMORY_EXCEEDED>()) {
+ // Only deal with non overcommit workload group.
+ if (wg->enable_memory_overcommit()) {
+ // Soft limit wg will only reserve failed when process
limit exceed. But in some corner case,
+ // when reserve, the wg is hard limit, the query reserve
failed, but when this loop run
+ // the wg is converted to soft limit.
+ // So that should resume the query.
+ LOG(WARNING)
+ << "Query: " << print_id(query_ctx->query_id())
+ << " reserve memory failed because exceed workload
group memlimit, it "
+ "should not happen, resume it again. paused
reason: "
+ << query_ctx->paused_reason();
+ query_ctx->set_memory_sufficient(true);
+ query_it = queries_list.erase(query_it);
+ continue;
+ }
+ // check if the reserve is too large, if it is too large,
+ // should set the query's limit only.
+ // Check the query's reserve with expected limit.
+ if (query_ctx->expected_mem_limit() <
+ query_ctx->get_mem_tracker()->consumption() +
query_it->reserve_size_) {
+ query_ctx->set_mem_limit(query_ctx->expected_mem_limit());
+ query_ctx->set_memory_sufficient(true);
+ LOG(INFO) << "Workload group memory reserve failed because
"
+ << query_ctx->debug_string() << " reserve size "
+ <<
PrettyPrinter::print_bytes(query_it->reserve_size_)
+ << " is too large, set hard limit to "
+ <<
PrettyPrinter::print_bytes(query_ctx->expected_mem_limit())
+ << " and resume running.";
+ query_it = queries_list.erase(query_it);
+ continue;
+ }
+ if (flushed_memtable_bytes <= 0) {
+ flushed_memtable_bytes =
+ flush_memtable_from_current_group_(wg,
query_it->reserve_size_);
+ }
+ if (flushed_memtable_bytes > 0) {
+ // Flushed some memtable, just wait flush finished and not
do anything more.
+ wg->enable_write_buffer_limit(true);
+ ++query_it;
+ continue;
+ }
+ if (!has_changed_hard_limit) {
+ update_queries_limit_(wg, true);
+ has_changed_hard_limit = true;
+ LOG(INFO) << "Query: " << print_id(query_ctx->query_id())
<< " reserve memory("
+ <<
PrettyPrinter::print_bytes(query_it->reserve_size_)
+ << ") failed due to workload group memory
exceed, "
+ "should set the workload group work in memory
insufficent mode, "
+ "so that other query will reduce their
memory. wg: "
+ << wg->debug_string();
+ }
+ if (wg->slot_memory_policy() == TWgSlotMemoryPolicy::NONE) {
+ // If not enable slot memory policy, then should spill
directly
+ // Maybe there are another query that use too much memory,
but we
+ // not encourage not enable slot memory.
+ // TODO should kill the query that exceed limit.
+ bool spill_res = handle_single_query_(query_ctx,
query_it->reserve_size_,
+
query_it->elapsed_time(),
+
query_ctx->paused_reason());
+ if (!spill_res) {
+ ++query_it;
+ continue;
+ } else {
+ query_it = queries_list.erase(query_it);
+ continue;
+ }
+ } else {
+ // Should not put the query back to task scheduler
immediately, because when wg's memory not sufficient,
+ // and then set wg's flag, other query may not free memory
very quickly.
+ if (query_it->elapsed_time() >
config::spill_in_paused_queue_timeout_ms) {
+ // set wg's memory to insufficent, then add it back to
task scheduler to run.
+ LOG(INFO) << "Query: " <<
print_id(query_ctx->query_id())
+ << " will be resume.";
+ query_ctx->set_memory_sufficient(true);
+ query_it = queries_list.erase(query_it);
+ continue;
+ } else {
+ ++query_it;
+ continue;
+ }
+ }
+ } else {
+ // If wg's memlimit not exceed, but process memory exceed, it
means cache or other metadata
+ // used too much memory. Should clean all cache here.
+ // 1. Check cache used, if cache is larger than > 0, then just
return and wait for it to 0 to release some memory.
+ if
(doris::GlobalMemoryArbitrator::last_affected_cache_capacity_adjust_weighted >
+ 0.001 &&
+
doris::GlobalMemoryArbitrator::last_wg_trigger_cache_capacity_adjust_weighted >
+ 0.001) {
+
doris::GlobalMemoryArbitrator::last_wg_trigger_cache_capacity_adjust_weighted =
+ 0;
+
doris::GlobalMemoryArbitrator::notify_cache_adjust_capacity();
+ LOG(INFO) << "There are some queries need process memory,
so that set cache "
+ "capacity "
+ "to 0 now";
+ }
+ if (query_it->cache_ratio_ < 0.001) {
+ // 1. Check if could revoke some memory from memtable
+ if (flushed_memtable_bytes <= 0) {
+ flushed_memtable_bytes =
+ flush_memtable_from_current_group_(wg,
query_it->reserve_size_);
+ }
+ if (flushed_memtable_bytes > 0) {
+ // Flushed some memtable, just wait flush finished and
not do anything more.
+ ++query_it;
+ continue;
+ }
+ // TODO should wait here to check if the process has
release revoked_size memory and then continue.
+ if (!has_revoked_from_other_group) {
+ int64_t revoked_size = revoke_memory_from_other_group_(
+ query_ctx, wg->enable_memory_overcommit(),
query_it->reserve_size_);
+ if (revoked_size > 0) {
+ has_revoked_from_other_group = true;
+ query_ctx->set_memory_sufficient(true);
+ query_it = queries_list.erase(query_it);
+ // Do not care if the revoked_size > reserve size,
and try to run again.
+ continue;
+ } else {
+ bool spill_res = handle_single_query_(
+ query_ctx, query_it->reserve_size_,
query_it->elapsed_time(),
+ query_ctx->paused_reason());
+ if (spill_res) {
+ query_it = queries_list.erase(query_it);
+ continue;
+ } else {
+ ++query_it;
+ continue;
+ }
+ }
+ } else {
+ // If any query is cancelled during process limit
stage, should resume other query and
+ // do not do any check now.
+ query_ctx->set_memory_sufficient(true);
+ query_it = queries_list.erase(query_it);
+ continue;
+ }
+ }
+ if
(doris::GlobalMemoryArbitrator::last_affected_cache_capacity_adjust_weighted <
+ 0.001 &&
+ query_it->cache_ratio_ > 0.001) {
+ LOG(INFO) << "Query: " << print_id(query_ctx->query_id())
+ << " will be resume after cache adjust.";
+ query_ctx->set_memory_sufficient(true);
+ query_it = queries_list.erase(query_it);
+ continue;
+ }
+ ++query_it;
+ }
+ }
+ // Not need waiting flush memtable and below low watermark disable
load buffer limit
+ if (flushed_memtable_bytes <= 0 && !is_low_watermark) {
+ wg->enable_write_buffer_limit(false);
+ }
+
+ if (queries_list.empty()) {
+ it = _paused_queries_list.erase(it);
+ continue;
+ } else {
+ // Finished deal with one workload group, and should deal with
next one.
+ ++it;
+ }
+ }
+}
+
+// Return the expected free bytes if memtable could flush
+int64_t WorkloadGroupMgr::flush_memtable_from_current_group_(WorkloadGroupPtr
wg,
+ int64_t
need_free_mem) {
+ // If there are a lot of memtable memory, then wait them flush finished.
+ MemTableMemoryLimiter* memtable_limiter =
+ doris::ExecEnv::GetInstance()->memtable_memory_limiter();
+ int64_t memtable_active_bytes = 0;
+ int64_t memtable_queue_bytes = 0;
+ int64_t memtable_flush_bytes = 0;
+ memtable_limiter->get_workload_group_memtable_usage(
+ wg->id(), &memtable_active_bytes, &memtable_queue_bytes,
&memtable_flush_bytes);
+ // TODO: should add a signal in memtable limiter to prevent new batch
+ // For example, streamload, it will not reserve many memory, but it will
occupy many memtable memory.
+ // TODO: 0.2 should be a workload group properties. For example, the group
is optimized for load,then the value
+ // should be larged, if the group is optimized for query, then the value
should be smaller.
+ int64_t max_wg_memtable_bytes = wg->write_buffer_limit();
+ if (memtable_active_bytes + memtable_queue_bytes + memtable_flush_bytes >
+ max_wg_memtable_bytes) {
+ // There are many table in flush queue, just waiting them flush
finished.
+ if (memtable_active_bytes < (int64_t)(max_wg_memtable_bytes * 0.6)) {
+ LOG_EVERY_T(INFO, 60) << wg->name()
+ << " load memtable size is: " <<
memtable_active_bytes << ", "
+ << memtable_queue_bytes << ", " <<
memtable_flush_bytes
+ << ", load buffer limit is: " <<
max_wg_memtable_bytes
+ << " wait for flush finished to release more
memory";
+ return memtable_queue_bytes + memtable_flush_bytes;
+ } else {
+ // Flush some memtables(currently written) to flush queue.
+ memtable_limiter->flush_workload_group_memtables(
+ wg->id(), memtable_active_bytes -
(int64_t)(max_wg_memtable_bytes * 0.6));
+ LOG_EVERY_T(INFO, 60) << wg->name()
+ << " load memtable size is: " <<
memtable_active_bytes << ", "
+ << memtable_queue_bytes << ", " <<
memtable_flush_bytes
+ << ", flush some active memtable to revoke
memory";
+ return memtable_queue_bytes + memtable_flush_bytes +
memtable_active_bytes -
+ (int64_t)(max_wg_memtable_bytes * 0.6);
+ }
+ }
+ return 0;
+}
+
+int64_t
WorkloadGroupMgr::revoke_memory_from_other_group_(std::shared_ptr<QueryContext>
requestor,
+ bool hard_limit,
int64_t need_free_mem) {
+ int64_t total_freed_mem = 0;
+ std::unique_ptr<RuntimeProfile> profile =
std::make_unique<RuntimeProfile>("RevokeMemory");
+ // 1. memtable like memory
+ // 2. query exceed workload group limit
+ int64_t freed_mem = revoke_overcommited_memory_(requestor, need_free_mem,
profile.get());
+ total_freed_mem += freed_mem;
+ // The revoke process may kill current requestor, so should return now.
+ if (need_free_mem - total_freed_mem < 0 || requestor->is_cancelled()) {
+ return total_freed_mem;
+ }
+ if (hard_limit) {
+ freed_mem = cancel_top_query_in_overcommit_group_(need_free_mem -
total_freed_mem,
+
doris::QUERY_MIN_MEMORY, profile.get());
+ } else {
+ freed_mem = cancel_top_query_in_overcommit_group_(
+ need_free_mem - total_freed_mem,
requestor->get_mem_tracker()->consumption(),
+ profile.get());
+ }
+ total_freed_mem += freed_mem;
+ // The revoke process may kill current requestor, so should return now.
+ if (need_free_mem - total_freed_mem < 0 || requestor->is_cancelled()) {
+ return total_freed_mem;
+ }
+ return total_freed_mem;
+}
+
+// Revoke memory from workload group that exceed it's limit. For example, if
the wg's limit is 10g, but used 12g
+// then should revoke 2g from the group.
+int64_t
WorkloadGroupMgr::revoke_overcommited_memory_(std::shared_ptr<QueryContext>
requestor,
+ int64_t need_free_mem,
+ RuntimeProfile* profile)
{
+ int64_t total_freed_mem = 0;
+ // 1. check memtable usage, and try to free them.
+ int64_t freed_mem =
revoke_memtable_from_overcommited_groups_(need_free_mem, profile);
+ total_freed_mem += freed_mem;
+ // The revoke process may kill current requestor, so should return now.
+ if (need_free_mem - total_freed_mem < 0 || requestor->is_cancelled()) {
+ return total_freed_mem;
+ }
+ // 2. Cancel top usage query, one by one
+ using WorkloadGroupMem = std::pair<WorkloadGroupPtr, int64_t>;
+ auto cmp = [](WorkloadGroupMem left, WorkloadGroupMem right) {
+ return left.second < right.second;
+ };
+ std::priority_queue<WorkloadGroupMem, std::vector<WorkloadGroupMem>,
decltype(cmp)> heap(cmp);
+ {
+ std::shared_lock<std::shared_mutex> r_lock(_group_mutex);
+ for (auto iter = _workload_groups.begin(); iter !=
_workload_groups.end(); iter++) {
+ if (requestor->workload_group() != nullptr &&
+ iter->second->id() == requestor->workload_group()->id()) {
+ continue;
+ }
+ heap.emplace(iter->second, iter->second->memory_used());
+ }
+ }
+ while (!heap.empty() && need_free_mem - total_freed_mem > 0 &&
!requestor->is_cancelled()) {
+ auto [wg, sort_mem] = heap.top();
+ heap.pop();
+ freed_mem = wg->free_overcommited_memory(need_free_mem -
total_freed_mem, profile);
+ total_freed_mem += freed_mem;
+ }
+ return total_freed_mem;
+}
+
+// If the memtable is too large, then flush them and wait for finished.
+int64_t WorkloadGroupMgr::revoke_memtable_from_overcommited_groups_(int64_t
need_free_mem,
+
RuntimeProfile* profile) {
+ return 0;
+}
+
+// 1. Sort all memory limiter in all overcommit wg, and cancel the top usage
task that with most memory.
+// 2. Maybe not valid because it's memory not exceed limit.
+int64_t WorkloadGroupMgr::cancel_top_query_in_overcommit_group_(int64_t
need_free_mem,
+ int64_t
lower_bound,
+
RuntimeProfile* profile) {
+ return 0;
+}
+
+// streamload, kafka routine load, group commit
+// insert into select
+// select
+
+// If the query could release some memory, for example, spill disk, then the
return value is true.
+// If the query could not release memory, then cancel the query, the return
value is true.
+// If the query is not ready to do these tasks, it means just wait, then
return value is false.
+bool WorkloadGroupMgr::handle_single_query_(const
std::shared_ptr<QueryContext>& query_ctx,
+ size_t size_to_reserve, int64_t
time_in_queue,
+ Status paused_reason) {
+ size_t revocable_size = 0;
+ size_t memory_usage = 0;
+ bool has_running_task = false;
+ const auto query_id = print_id(query_ctx->query_id());
+ query_ctx->get_revocable_info(&revocable_size, &memory_usage,
&has_running_task);
+ if (has_running_task) {
+ LOG(INFO) << "Query: " << print_id(query_ctx->query_id())
+ << " is paused, but still has running task, skip it.";
+ return false;
+ }
+
+ auto revocable_tasks = query_ctx->get_revocable_tasks();
+ if (revocable_tasks.empty()) {
+ if (paused_reason.is<ErrorCode::QUERY_MEMORY_EXCEEDED>()) {
+ const auto limit = query_ctx->get_mem_limit();
+ const auto reserved_size =
query_ctx->query_mem_tracker->reserved_consumption();
+ // During waiting time, another operator in the query may finished
and release
+ // many memory and we could run.
+ if ((memory_usage + size_to_reserve) < limit) {
+ LOG(INFO) << "Query: " << query_id << ", usage("
+ << PrettyPrinter::print_bytes(memory_usage) << " + "
<< size_to_reserve
+ << ") less than limit(" <<
PrettyPrinter::print_bytes(limit)
+ << "), resume it.";
+ query_ctx->set_memory_sufficient(true);
+ return true;
+ } else if (time_in_queue >=
config::spill_in_paused_queue_timeout_ms) {
+ // Use MEM_LIMIT_EXCEEDED so that FE could parse the error
code and do try logic
+ auto msg1 = fmt::format(
+ "Query {} reserve memory failed, but could not find
memory that could "
+ "release or spill to disk. Query memory usage: {},
reserved size: {}, try "
+ "to reserve: {} "
+ ", limit: {} ,process memory info: {}, wg info: {}.",
+ query_id, PrettyPrinter::print_bytes(memory_usage),
+ PrettyPrinter::print_bytes(reserved_size),
+ PrettyPrinter::print_bytes(size_to_reserve),
+ PrettyPrinter::print_bytes(query_ctx->get_mem_limit()),
+
GlobalMemoryArbitrator::process_memory_used_details_str(),
+ query_ctx->workload_group()->memory_debug_string());
+ auto msg2 = msg1 + fmt::format(
+ " Query Memory Tracker Summary: {}."
+ " Load Memory Tracker Summary: {}",
+
MemTrackerLimiter::make_type_trackers_profile_str(
+
MemTrackerLimiter::Type::QUERY),
+
MemTrackerLimiter::make_type_trackers_profile_str(
+
MemTrackerLimiter::Type::LOAD));
+ LOG(INFO) << msg2;
+
query_ctx->cancel(doris::Status::Error<ErrorCode::MEM_LIMIT_EXCEEDED>(msg1));
+ } else {
+ return false;
+ }
+ } else if
(paused_reason.is<ErrorCode::WORKLOAD_GROUP_MEMORY_EXCEEDED>()) {
+ if (!query_ctx->workload_group()->exceed_limit()) {
+ LOG(INFO) << "Query: " << query_id
+ << " paused caused by
WORKLOAD_GROUP_MEMORY_EXCEEDED, now resume it.";
+ query_ctx->set_memory_sufficient(true);
+ return true;
+ } else if (time_in_queue >
config::spill_in_paused_queue_timeout_ms) {
+ LOG(INFO) << "Query: " << query_id << ", workload group
exceeded, info: "
+ <<
GlobalMemoryArbitrator::process_memory_used_details_str()
+ << ", wg info: " <<
query_ctx->workload_group()->memory_debug_string();
+
query_ctx->cancel(doris::Status::Error<ErrorCode::MEM_LIMIT_EXCEEDED>(
+ "The query({}) reserved memory failed because workload
group limit "
+ "exceeded, and there is no cache now. And could not
find task to spill. "
+ "Maybe you should set the workload group's limit to a
lower value.",
+ query_id));
+ } else {
+ return false;
+ }
+ } else {
+ // Should not consider about process memory. For example, the
query's limit is 100g, workload
+ // group's memlimit is 10g, process memory is 20g. The query
reserve will always failed in wg
+ // limit, and process is always have memory, so that it will
resume and failed reserve again.
+ if (!GlobalMemoryArbitrator::is_exceed_hard_mem_limit()) {
+ LOG(INFO) << "Query: " << query_id
+ << ", process limit not exceeded now, resume this
query"
+ << ", process memory info: "
+ <<
GlobalMemoryArbitrator::process_memory_used_details_str()
+ << ", wg info: " <<
query_ctx->workload_group()->memory_debug_string();
+ query_ctx->set_memory_sufficient(true);
+ return true;
+ } else if (time_in_queue >
config::spill_in_paused_queue_timeout_ms) {
+ LOG(INFO) << "Query: " << query_id << ", process limit
exceeded, info: "
+ <<
GlobalMemoryArbitrator::process_memory_used_details_str()
+ << ", wg info: " <<
query_ctx->workload_group()->memory_debug_string();
+
query_ctx->cancel(doris::Status::Error<ErrorCode::MEM_LIMIT_EXCEEDED>(
+ "The query({}) reserved memory failed because process
limit exceeded, "
+ "and "
+ "there is no cache now. And could not find task to
spill. Maybe you "
+ "should "
+ "set "
+ "the workload group's limit to a lower value.",
+ query_id));
+ } else {
+ return false;
+ }
+ }
+ } else {
+ SCOPED_ATTACH_TASK(query_ctx.get());
+ auto status = query_ctx->revoke_memory();
+ if (!status.ok()) {
+
ExecEnv::GetInstance()->fragment_mgr()->cancel_query(query_ctx->query_id(),
status);
+ }
+ }
+ return true;
+}
+
+void WorkloadGroupMgr::update_queries_limit_(WorkloadGroupPtr wg, bool
enable_hard_limit) {
Review Comment:
warning: function 'update_queries_limit_' exceeds recommended
size/complexity thresholds [readability-function-size]
```cpp
void WorkloadGroupMgr::update_queries_limit_(WorkloadGroupPtr wg, bool
enable_hard_limit) {
^
```
<details>
<summary>Additional context</summary>
**be/src/runtime/workload_group/workload_group_manager.cpp:754:** 98 lines
including whitespace and comments (threshold 80)
```cpp
void WorkloadGroupMgr::update_queries_limit_(WorkloadGroupPtr wg, bool
enable_hard_limit) {
^
```
</details>
##########
be/src/pipeline/exec/partitioned_aggregation_source_operator.cpp:
##########
@@ -190,109 +235,112 @@ Status
PartitionedAggLocalState::setup_in_memory_agg_op(RuntimeState* state) {
return source_local_state->open(state);
}
-Status
PartitionedAggLocalState::initiate_merge_spill_partition_agg_data(RuntimeState*
state) {
- DCHECK(!_is_merging);
-
Base::_shared_state->in_mem_shared_state->aggregate_data_container->init_once();
- if
(Base::_shared_state->in_mem_shared_state->aggregate_data_container->iterator !=
-
Base::_shared_state->in_mem_shared_state->aggregate_data_container->end() ||
- _shared_state->spill_partitions.empty()) {
+Status PartitionedAggLocalState::recover_blocks_from_disk(RuntimeState* state,
bool& has_data) {
Review Comment:
warning: function 'recover_blocks_from_disk' has cognitive complexity of 76
(threshold 50) [readability-function-cognitive-complexity]
```cpp
Status PartitionedAggLocalState::recover_blocks_from_disk(RuntimeState*
state, bool& has_data) {
^
```
<details>
<summary>Additional context</summary>
**be/src/pipeline/exec/partitioned_aggregation_source_operator.cpp:240:**
+1, including nesting penalty of 0, nesting level increased to 1
```cpp
if (_shared_state->spill_partitions.empty()) {
^
```
**be/src/pipeline/exec/partitioned_aggregation_source_operator.cpp:247:**
nesting level increased to 1
```cpp
auto spill_func = [this, state, query_id] {
^
```
**be/src/pipeline/exec/partitioned_aggregation_source_operator.cpp:260:**
+2, including nesting penalty of 1, nesting level increased to 2
```cpp
while (!state->is_cancelled() && !has_agg_data &&
^
```
**be/src/pipeline/exec/partitioned_aggregation_source_operator.cpp:262:**
+3, including nesting penalty of 2, nesting level increased to 3
```cpp
while
(!_shared_state->spill_partitions[0]->spill_streams_.empty() &&
^
```
**be/src/pipeline/exec/partitioned_aggregation_source_operator.cpp:268:**
+4, including nesting penalty of 3, nesting level increased to 4
```cpp
while (!eos && !state->is_cancelled()) {
^
```
**be/src/pipeline/exec/partitioned_aggregation_source_operator.cpp:268:** +1
```cpp
while (!eos && !state->is_cancelled()) {
^
```
**be/src/pipeline/exec/partitioned_aggregation_source_operator.cpp:270:**
+5, including nesting penalty of 4, nesting level increased to 5
```cpp
DBUG_EXECUTE_IF("fault_inject::partitioned_agg_source::recover_spill_data",
^
```
**be/src/util/debug_points.h:36:** expanded from macro 'DBUG_EXECUTE_IF'
```cpp
if (UNLIKELY(config::enable_debug_points)) {
\
^
```
**be/src/pipeline/exec/partitioned_aggregation_source_operator.cpp:270:**
+6, including nesting penalty of 5, nesting level increased to 6
```cpp
DBUG_EXECUTE_IF("fault_inject::partitioned_agg_source::recover_spill_data",
^
```
**be/src/util/debug_points.h:38:** expanded from macro 'DBUG_EXECUTE_IF'
```cpp
if (dp) {
\
^
```
**be/src/pipeline/exec/partitioned_aggregation_source_operator.cpp:276:**
+5, including nesting penalty of 4, nesting level increased to 5
```cpp
if (status.ok()) {
^
```
**be/src/pipeline/exec/partitioned_aggregation_source_operator.cpp:280:**
+5, including nesting penalty of 4, nesting level increased to 5
```cpp
RETURN_IF_ERROR(status);
^
```
**be/src/common/status.h:642:** expanded from macro 'RETURN_IF_ERROR'
```cpp
do { \
^
```
**be/src/pipeline/exec/partitioned_aggregation_source_operator.cpp:280:**
+6, including nesting penalty of 5, nesting level increased to 6
```cpp
RETURN_IF_ERROR(status);
^
```
**be/src/common/status.h:644:** expanded from macro 'RETURN_IF_ERROR'
```cpp
if (UNLIKELY(!_status_.ok())) { \
^
```
**be/src/pipeline/exec/partitioned_aggregation_source_operator.cpp:282:**
+5, including nesting penalty of 4, nesting level increased to 5
```cpp
if (!block.empty()) {
^
```
**be/src/pipeline/exec/partitioned_aggregation_source_operator.cpp:287:**
+6, including nesting penalty of 5, nesting level increased to 6
```cpp
if (accumulated_blocks_size >=
^
```
**be/src/pipeline/exec/partitioned_aggregation_source_operator.cpp:296:**
+4, including nesting penalty of 3, nesting level increased to 4
```cpp
if (_current_partition_eos) {
^
```
**be/src/pipeline/exec/partitioned_aggregation_source_operator.cpp:302:**
+3, including nesting penalty of 2, nesting level increased to 3
```cpp
if (_shared_state->spill_partitions[0]->spill_streams_.empty()) {
^
```
**be/src/pipeline/exec/partitioned_aggregation_source_operator.cpp:315:**
nesting level increased to 1
```cpp
auto exception_catch_func = [spill_func, query_id]() {
^
```
**be/src/pipeline/exec/partitioned_aggregation_source_operator.cpp:316:**
+2, including nesting penalty of 1, nesting level increased to 2
```cpp
DBUG_EXECUTE_IF("fault_inject::partitioned_agg_source::merge_spill_data_cancel",
{
^
```
**be/src/util/debug_points.h:36:** expanded from macro 'DBUG_EXECUTE_IF'
```cpp
if (UNLIKELY(config::enable_debug_points)) {
\
^
```
**be/src/pipeline/exec/partitioned_aggregation_source_operator.cpp:316:**
+3, including nesting penalty of 2, nesting level increased to 3
```cpp
DBUG_EXECUTE_IF("fault_inject::partitioned_agg_source::merge_spill_data_cancel",
{
^
```
**be/src/util/debug_points.h:38:** expanded from macro 'DBUG_EXECUTE_IF'
```cpp
if (dp) {
\
^
```
**be/src/pipeline/exec/partitioned_aggregation_source_operator.cpp:324:**
nesting level increased to 2
```cpp
auto status = [&]() { RETURN_IF_CATCH_EXCEPTION({ return
spill_func(); }); }();
^
```
**be/src/pipeline/exec/partitioned_aggregation_source_operator.cpp:324:**
+3, including nesting penalty of 2, nesting level increased to 3
```cpp
auto status = [&]() { RETURN_IF_CATCH_EXCEPTION({ return
spill_func(); }); }();
^
```
**be/src/common/exception.h:79:** expanded from macro
'RETURN_IF_CATCH_EXCEPTION'
```cpp
do {
\
^
```
**be/src/pipeline/exec/partitioned_aggregation_source_operator.cpp:324:**
+4, including nesting penalty of 3, nesting level increased to 4
```cpp
auto status = [&]() { RETURN_IF_CATCH_EXCEPTION({ return
spill_func(); }); }();
^
```
**be/src/common/exception.h:84:** expanded from macro
'RETURN_IF_CATCH_EXCEPTION'
```cpp
} catch (const doris::Exception& e) {
\
^
```
**be/src/pipeline/exec/partitioned_aggregation_source_operator.cpp:324:**
+5, including nesting penalty of 4, nesting level increased to 5
```cpp
auto status = [&]() { RETURN_IF_CATCH_EXCEPTION({ return
spill_func(); }); }();
^
```
**be/src/common/exception.h:85:** expanded from macro
'RETURN_IF_CATCH_EXCEPTION'
```cpp
if (e.code() == doris::ErrorCode::MEM_ALLOC_FAILED) {
\
^
```
**be/src/pipeline/exec/partitioned_aggregation_source_operator.cpp:330:**
+1, including nesting penalty of 0, nesting level increased to 1
```cpp
DBUG_EXECUTE_IF("fault_inject::partitioned_agg_source::submit_func", {
^
```
**be/src/util/debug_points.h:36:** expanded from macro 'DBUG_EXECUTE_IF'
```cpp
if (UNLIKELY(config::enable_debug_points)) {
\
^
```
**be/src/pipeline/exec/partitioned_aggregation_source_operator.cpp:330:**
+2, including nesting penalty of 1, nesting level increased to 2
```cpp
DBUG_EXECUTE_IF("fault_inject::partitioned_agg_source::submit_func", {
^
```
**be/src/util/debug_points.h:38:** expanded from macro 'DBUG_EXECUTE_IF'
```cpp
if (dp) {
\
^
```
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
