[feat] support rf in partitioned hash join #45637
Conversation
|
Thank you for your contribution to Apache Doris. Please clearly describe your PR:
|
![github-actions[bot]](https://avatars.githubusercontent.com/in/15368?s=60&v=4){kind=small}
be/src/exprs/runtime_filter.cpp
Outdated
| @@ -476,6 +476,18 @@ class RuntimePredicateWrapper { | |||
| const TExpr& probe_expr); | |||
|
|
|||
| Status merge(const RuntimePredicateWrapper* wrapper) { | |||
There was a problem hiding this comment.
warning: function 'merge' has cognitive complexity of 78 (threshold 50) [readability-function-cognitive-complexity]
Status merge(const RuntimePredicateWrapper* wrapper) {
^Additional context
be/src/exprs/runtime_filter.cpp:478: +1, including nesting penalty of 0, nesting level increased to 1
if (wrapper->_context->disabled) {
^be/src/exprs/runtime_filter.cpp:486: +1, including nesting penalty of 0, nesting level increased to 1
if (_context->disabled) {
^be/src/exprs/runtime_filter.cpp:490: +1, including nesting penalty of 0, nesting level increased to 1
if (wrapper->is_ignored()) {
^be/src/exprs/runtime_filter.cpp:496: +1
_filter_type == RuntimeFilterType::IN_OR_BLOOM_FILTER &&
^be/src/exprs/runtime_filter.cpp:501: +1
bool can_not_merge_other = _filter_type != RuntimeFilterType::IN_OR_BLOOM_FILTER &&
^be/src/exprs/runtime_filter.cpp:504: +1
CHECK(!can_not_merge_in_or_bloom && !can_not_merge_other)
^be/src/exprs/runtime_filter.cpp:509: +1, including nesting penalty of 0, nesting level increased to 1
switch (_filter_type) {
^be/src/exprs/runtime_filter.cpp:511: +2, including nesting penalty of 1, nesting level increased to 2
if (!_context->hybrid_set) {
^be/src/exprs/runtime_filter.cpp:516: +2, including nesting penalty of 1, nesting level increased to 2
if (_max_in_num >= 0 && _context->hybrid_set->size() >= _max_in_num) {
^be/src/exprs/runtime_filter.cpp:526: +2, including nesting penalty of 1, nesting level increased to 2
RETURN_IF_ERROR(_context->minmax_func->merge(wrapper->_context->minmax_func.get()));
^be/src/common/status.h:642: expanded from macro 'RETURN_IF_ERROR'
do { \
^be/src/exprs/runtime_filter.cpp:526: +3, including nesting penalty of 2, nesting level increased to 3
RETURN_IF_ERROR(_context->minmax_func->merge(wrapper->_context->minmax_func.get()));
^be/src/common/status.h:644: expanded from macro 'RETURN_IF_ERROR'
if (UNLIKELY(!_status_.ok())) { \
^be/src/exprs/runtime_filter.cpp:530: +2, including nesting penalty of 1, nesting level increased to 2
RETURN_IF_ERROR(
^be/src/common/status.h:642: expanded from macro 'RETURN_IF_ERROR'
do { \
^be/src/exprs/runtime_filter.cpp:530: +3, including nesting penalty of 2, nesting level increased to 3
RETURN_IF_ERROR(
^be/src/common/status.h:644: expanded from macro 'RETURN_IF_ERROR'
if (UNLIKELY(!_status_.ok())) { \
^be/src/exprs/runtime_filter.cpp:538: +2, including nesting penalty of 1, nesting level increased to 2
if (other_filter_type == RuntimeFilterType::IN_OR_BLOOM_FILTER) {
^be/src/exprs/runtime_filter.cpp:542: +2, including nesting penalty of 1, nesting level increased to 2
if (real_filter_type == RuntimeFilterType::IN_FILTER) {
^be/src/exprs/runtime_filter.cpp:546: +3, including nesting penalty of 2, nesting level increased to 3
if (other_filter_type == RuntimeFilterType::IN_FILTER) {
^be/src/exprs/runtime_filter.cpp:548: +4, including nesting penalty of 3, nesting level increased to 4
if (_max_in_num >= 0 && _context->hybrid_set->size() >= _max_in_num) {
^be/src/exprs/runtime_filter.cpp:550: +5, including nesting penalty of 4, nesting level increased to 5
RETURN_IF_ERROR(_context->bloom_filter_func->init_with_fixed_length());
^be/src/common/status.h:642: expanded from macro 'RETURN_IF_ERROR'
do { \
^be/src/exprs/runtime_filter.cpp:550: +6, including nesting penalty of 5, nesting level increased to 6
RETURN_IF_ERROR(_context->bloom_filter_func->init_with_fixed_length());
^be/src/common/status.h:644: expanded from macro 'RETURN_IF_ERROR'
if (UNLIKELY(!_status_.ok())) { \
^be/src/exprs/runtime_filter.cpp:551: +5, including nesting penalty of 4, nesting level increased to 5
RETURN_IF_ERROR(change_to_bloom_filter());
^be/src/common/status.h:642: expanded from macro 'RETURN_IF_ERROR'
do { \
^be/src/exprs/runtime_filter.cpp:551: +6, including nesting penalty of 5, nesting level increased to 6
RETURN_IF_ERROR(change_to_bloom_filter());
^be/src/common/status.h:644: expanded from macro 'RETURN_IF_ERROR'
if (UNLIKELY(!_status_.ok())) { \
^be/src/exprs/runtime_filter.cpp:553: +1, nesting level increased to 3
} else {
^be/src/exprs/runtime_filter.cpp:556: +4, including nesting penalty of 3, nesting level increased to 4
RETURN_IF_ERROR(change_to_bloom_filter());
^be/src/common/status.h:642: expanded from macro 'RETURN_IF_ERROR'
do { \
^be/src/exprs/runtime_filter.cpp:556: +5, including nesting penalty of 4, nesting level increased to 5
RETURN_IF_ERROR(change_to_bloom_filter());
^be/src/common/status.h:644: expanded from macro 'RETURN_IF_ERROR'
if (UNLIKELY(!_status_.ok())) { \
^be/src/exprs/runtime_filter.cpp:558: +1, nesting level increased to 2
} else {
^be/src/exprs/runtime_filter.cpp:559: +3, including nesting penalty of 2, nesting level increased to 3
if (other_filter_type == RuntimeFilterType::IN_FILTER) {
^be/src/exprs/runtime_filter.cpp:562: +1, nesting level increased to 3
} else {
^be/src/exprs/runtime_filter.cpp:564: +4, including nesting penalty of 3, nesting level increased to 4
RETURN_IF_ERROR(_context->bloom_filter_func->merge(
^be/src/common/status.h:642: expanded from macro 'RETURN_IF_ERROR'
do { \
^be/src/exprs/runtime_filter.cpp:564: +5, including nesting penalty of 4, nesting level increased to 5
RETURN_IF_ERROR(_context->bloom_filter_func->merge(
^be/src/common/status.h:644: expanded from macro 'RETURN_IF_ERROR'
if (UNLIKELY(!_status_.ok())) { \
^
| @@ -86,6 +106,46 @@ Status PartitionedHashJoinSinkLocalState::close(RuntimeState* state, Status exec | |||
| return Status::OK(); | |||
| } | |||
| dec_running_big_mem_op_num(state); | |||
|
|
|||
| if (_runtime_filters.empty() || _shared_state->need_to_spill) { | |||
| return PipelineXSpillSinkLocalState::close(state, exec_status); | |||
There was a problem hiding this comment.
InMemoryHashJoinOperator
disable runtime filter（）
sink（eos=true）
close()
PartitionedHashJoinOperator(){
first partition():disable runtime filter();
}
mrhhsg
force-pushed
the
spill_and_reserve
branch
from
65dd428 to
7d4a581
Compare
What problem does this PR solve?
Issue Number: close #xxx
Related PR: #xxx
Problem Summary:
Release note
None
Check List (For Author)
Test
Behavior changed:
Does this need documentation?
Check List (For Reviewer who merge this PR)