This is an automated email from the ASF dual-hosted git repository. BiteTheDDDDt pushed a commit to branch opt_perf_4.1 in repository https://gitbox.apache.org/repos/asf/doris.git
commit 4a15f2dd63fb7b489e1a961695abb7bc012f48b7 Author: Pxl <[email protected]> AuthorDate: Tue Jul 14 11:42:51 2026 +0800 [improvement](be) Optimize hash join probe side output (#65300) Hash join probe-side output may still copy projected probe columns even when the output rows map one-to-one to the original probe block. In that ALL_MATCH_ONE case, `_probe_indexs` covers the whole probe block in order, so copying probe-side columns through `insert_range_from()` is unnecessary. This PR adds a zero-copy fast path for eligible probe-side output columns: - Detect batches where `_probe_indexs` is `0..probe_block.rows()-1` and output row count equals the probe block row count. - Keep the mutable output block shape by mocking eligible probe-side columns during output construction. - After materializing the output `Block`, replace eligible probe-side output columns with the original probe block columns. - Limit the fast path to join types where the probe-side output type is unchanged: inner join, left outer join, left semi join, left anti join, ASOF left inner join, and ASOF left outer join. - Keep conservative guards for cases that can change row filtering or column materialization: other join conjuncts, mark join conjuncts, local conjuncts, variant finalization, right/full joins, and null-aware joins. Local TPC-DS SF100 q47 profile data on a single-BE release build: Query settings: ```sql set enable_profile=true; set profile_level=2; set disable_join_reorder=false; ``` End-to-end elapsed time is noisy and did not show a stable query-level improvement because build-side output and projection still dominate and pipeline overlap hides part of the local CPU saving. | Run | Query/Profile id | q47 elapsed | Output check | | --- | --- | ---: | --- | | Baseline run 1 | N/A | 7.31s | baseline output | | Baseline run 2 | `a3e69b5226774770-a08c4bf2539526ce` | 6.42s | baseline output | | After this PR | `dfeb8b9ab13546a5-820c7ea81e3cbe23` | 6.93s | identical to baseline | Focused hash join counters show the intended reduction in probe-side output time: | Operator | Counter | Baseline | After this PR | Change | | --- | --- | ---: | ---: | ---: | | HASH_JOIN_OPERATOR id=9 | `ProbeWhenProbeSideOutputTime` | 481.862ms | 52.714ms | -89.1% | | HASH_JOIN_OPERATOR id=9 | `ProbeWhenBuildSideOutputTime` | 1sec779ms | 1sec858ms | noise / unchanged | | HASH_JOIN_OPERATOR id=9 | `ProbeWhenSearchHashTableTime` | 229.461ms | 232.211ms | noise / unchanged | | HASH_JOIN_OPERATOR id=9 | `ProjectionTime` | 667.907ms | 820.823ms | noise | | HASH_JOIN_OPERATOR id=9 | `ExecTime` | 3sec435ms | 3sec228ms | -6.0% | | HASH_JOIN_OPERATOR id=7 | `ProbeWhenProbeSideOutputTime` | 275.201ms | 41.642ms | -84.9% | | HASH_JOIN_OPERATOR id=7 | `ProbeWhenBuildSideOutputTime` | 94.786ms | 110.221ms | noise / unchanged | | HASH_JOIN_OPERATOR id=7 | `ExecTime` | 717.242ms | 546.437ms | -23.8% | | HASH_JOIN_OPERATOR id=8 | `ProbeWhenProbeSideOutputTime` | 327.954ms | 394.574ms | not eligible / noise | The q47 result file from the final run was byte-identical to the baseline result file. ### Release note None ### Check List (For Author) - Test <!-- At least one of them must be included. --> - [ ] Regression test - [ ] Unit Test - [x] Manual test (add detailed scripts or steps below) - `build-support/clang-format.sh be/src/exec/operator/join/process_hash_table_probe.h be/src/exec/operator/join/process_hash_table_probe_impl.h` - `git diff --check` - `./build.sh --be` - TPC-DS SF100 q47 on a local single-BE release cluster with the settings above; compared final output with the baseline output. - Tried `build-support/run-clang-tidy.sh --base upstream/master --build-dir be/build_Release`; it failed on existing large-function/complexity diagnostics in `process_hash_table_probe_impl.h` and static assertions from included `jni-util.h`, not on the new changed lines. - [ ] No need to test or manual test. Explain why: - [ ] This is a refactor/code format and no logic has been changed. - [ ] Previous test can cover this change. - [ ] No code files have been changed. - [ ] Other reason <!-- Add your reason? --> - Behavior changed: - [x] No. - [ ] Yes. <!-- Explain the behavior change --> - Does this need documentation? - [x] No. - [ ] Yes. <!-- Add document PR link here. eg: https://github.com/apache/doris-website/pull/1214 --> ### Check List (For Reviewer who merge this PR) - [ ] Confirm the release note - [ ] Confirm test cases - [ ] Confirm document - [ ] Add branch pick label <!-- Add branch pick label that this PR should merge into --> (cherry picked from commit 4809030bd52ffea587254a3b01746e91f755ee61) --- .../exec/operator/join/process_hash_table_probe.h | 4 +- .../operator/join/process_hash_table_probe_impl.h | 74 ++++++++++---- .../join/test_hash_join_probe_side_zero_copy.out | 16 +++ .../test_hash_join_probe_side_zero_copy.groovy | 111 +++++++++++++++++++++ 4 files changed, 184 insertions(+), 21 deletions(-) diff --git a/be/src/exec/operator/join/process_hash_table_probe.h b/be/src/exec/operator/join/process_hash_table_probe.h index c5b86951d18..f17e49d448f 100644 --- a/be/src/exec/operator/join/process_hash_table_probe.h +++ b/be/src/exec/operator/join/process_hash_table_probe.h @@ -47,6 +47,7 @@ struct ProcessHashTableProbe { void build_side_output_column(MutableColumns& mcol, bool is_mark_join); void probe_side_output_column(MutableColumns& mcol); + bool can_transfer_probe_columns_to_output(bool probe_indices_are_contiguous) const; // Only process the join with no other join conjunct, because of no other join conjunt // the output block struct is same with mutable block. we can do more opt on it and simplify @@ -120,7 +121,8 @@ struct ProcessHashTableProbe { // nullable column but not has null except first row std::vector<bool> _build_column_has_null; - bool _need_calculate_all_match_one = false; + bool _should_check_probe_index_continuity = false; + bool _can_transfer_probe_columns_to_output = false; RuntimeProfile::Counter* _search_hashtable_timer = nullptr; RuntimeProfile::Counter* _init_probe_side_timer = nullptr; diff --git a/be/src/exec/operator/join/process_hash_table_probe_impl.h b/be/src/exec/operator/join/process_hash_table_probe_impl.h index 3b28eaa5b8e..11bc670db09 100644 --- a/be/src/exec/operator/join/process_hash_table_probe_impl.h +++ b/be/src/exec/operator/join/process_hash_table_probe_impl.h @@ -34,24 +34,25 @@ namespace doris { #include "common/compile_check_begin.h" -static bool check_all_match_one(const auto& vecs) { - size_t size = vecs.size(); - if (!size || vecs[size - 1] != vecs[0] + size - 1) { +static bool are_indices_contiguous(const auto& indices) { + size_t size = indices.size(); + if (!size || indices[size - 1] != indices[0] + size - 1) { return false; } for (size_t i = 1; i < size; i++) { - if (vecs[i] == vecs[i - 1]) { + if (indices[i] == indices[i - 1]) { return false; } } return true; } -static void insert_with_indexs(auto& dst, const auto& src, const auto& indexs, bool all_match_one) { - if (all_match_one) { - dst->insert_range_from(*src, indexs[0], indexs.size()); +static void insert_with_indices(auto& dst, const auto& src, const auto& indices, + bool indices_are_contiguous) { + if (indices_are_contiguous) { + dst->insert_range_from(*src, indices[0], indices.size()); } else { - dst->insert_indices_from(*src, indexs.data(), indexs.data() + indexs.size()); + dst->insert_indices_from(*src, indices.data(), indices.data() + indices.size()); } } @@ -85,15 +86,15 @@ ProcessHashTableProbe<JoinOpType>::ProcessHashTableProbe(HashJoinProbeLocalState _asof_probe_search_timer(parent->_asof_probe_search_timer), _right_col_idx(_parent_operator->_right_col_idx), _right_col_len(_parent_operator->_right_table_data_types.size()) { - constexpr int CALCULATE_ALL_MATCH_ONE_THRESHOLD = 2; + constexpr int CONTIGUOUS_INDICES_CHECK_THRESHOLD = 2; int probe_output_non_lazy_materialized_count = 0; for (int i = 0; i < _left_output_slot_flags.size(); i++) { if (_left_output_slot_flags[i] && !_parent_operator->is_lazy_materialized_column(i)) { probe_output_non_lazy_materialized_count++; } } - _need_calculate_all_match_one = - probe_output_non_lazy_materialized_count >= CALCULATE_ALL_MATCH_ONE_THRESHOLD; + _should_check_probe_index_continuity = + probe_output_non_lazy_materialized_count >= CONTIGUOUS_INDICES_CHECK_THRESHOLD; } template <int JoinOpType> @@ -155,12 +156,35 @@ void ProcessHashTableProbe<JoinOpType>::build_side_output_column(MutableColumns& } } +template <int JoinOpType> +bool ProcessHashTableProbe<JoinOpType>::can_transfer_probe_columns_to_output( + bool probe_indices_are_contiguous) const { + constexpr bool probe_side_type_unchanged = + JoinOpType == TJoinOp::INNER_JOIN || JoinOpType == TJoinOp::LEFT_OUTER_JOIN || + JoinOpType == TJoinOp::LEFT_SEMI_JOIN || JoinOpType == TJoinOp::LEFT_ANTI_JOIN || + JoinOpType == TJoinOp::ASOF_LEFT_INNER_JOIN || + JoinOpType == TJoinOp::ASOF_LEFT_OUTER_JOIN; + if constexpr (!probe_side_type_unchanged) { + return false; + } + + // Only transfer ownership after the whole probe block has been consumed and there is no + // pending build-side match that will need the original probe columns in a later pull. + return probe_indices_are_contiguous && _probe_indexs.get_element(0) == 0 && + _probe_indexs.size() == _parent->_probe_block.rows() && + _parent->_probe_index == _parent->_probe_block.rows() && _parent->_build_index == 0 && + !_parent_operator->need_finalize_variant_column(); +} + template <int JoinOpType> void ProcessHashTableProbe<JoinOpType>::probe_side_output_column(MutableColumns& mcol) { SCOPED_TIMER(_probe_side_output_timer); auto& probe_block = _parent->_probe_block; - bool all_match_one = - _need_calculate_all_match_one ? check_all_match_one(_probe_indexs.get_data()) : false; + bool probe_indices_are_contiguous = _should_check_probe_index_continuity + ? are_indices_contiguous(_probe_indexs.get_data()) + : false; + _can_transfer_probe_columns_to_output = + can_transfer_probe_columns_to_output(probe_indices_are_contiguous); for (int i = 0; i < _left_output_slot_flags.size(); ++i) { if (_left_output_slot_flags[i]) { @@ -177,8 +201,18 @@ void ProcessHashTableProbe<JoinOpType>::probe_side_output_column(MutableColumns& bool should_output = _left_output_slot_flags[i] && (is_asof_join || !_parent_operator->is_lazy_materialized_column(i)); if (should_output) { - auto& column = probe_block.get_by_position(i).column; - insert_with_indexs(mcol[i], column, _probe_indexs.get_data(), all_match_one); + if (_can_transfer_probe_columns_to_output) { + auto& probe_column = probe_block.get_by_position(i).column; + auto mutable_probe_column = IColumn::mutate(std::move(probe_column)); + // Reuse the empty output column as the next child output column. The transferred + // probe column has been fully consumed and no longer needs to carry the row count. + probe_column = std::move(mcol[i]); + mcol[i] = std::move(mutable_probe_column); + } else { + auto& column = probe_block.get_by_position(i).column; + insert_with_indices(mcol[i], column, _probe_indexs.get_data(), + probe_indices_are_contiguous); + } } else { mock_column_size(mcol[i], _probe_indexs.size()); } @@ -467,9 +501,9 @@ void ProcessHashTableProbe<JoinOpType>::process_direct_return(HashTableType& has probe_indexs_data[i] = i; } auto& mcol = mutable_block.mutable_columns(); + _parent->_probe_index = probe_rows; probe_side_output_column(mcol); output_block->swap(mutable_block.to_block()); - _parent->_probe_index = probe_rows; } template <int JoinOpType> @@ -629,7 +663,7 @@ Status ProcessHashTableProbe<JoinOpType>::finalize_block_with_filter(Block* outp auto do_lazy_materialize = [&](const std::vector<bool>& output_slot_flags, ColumnOffset32& row_indexs, int column_offset, - Block* source_block, bool try_all_match_one) { + Block* source_block, bool check_index_continuity) { std::vector<int> column_ids; for (int i = 0; i < output_slot_flags.size(); ++i) { if (output_slot_flags[i] && @@ -647,7 +681,7 @@ Status ProcessHashTableProbe<JoinOpType>::finalize_block_with_filter(Block* outp } const auto& container = row_indexs.get_data(); - bool all_match_one = try_all_match_one && check_all_match_one(container); + bool indices_are_contiguous = check_index_continuity && are_indices_contiguous(container); for (int column_id : column_ids) { int output_column_id = column_id + column_offset; output_block->get_by_position(output_column_id).column = @@ -659,13 +693,13 @@ Status ProcessHashTableProbe<JoinOpType>::finalize_block_with_filter(Block* outp auto dst = IColumn::mutate( std::move(output_block->get_by_position(output_column_id).column)); dst->clear(); - insert_with_indexs(dst, src, container, all_match_one); + insert_with_indices(dst, src, container, indices_are_contiguous); output_block->get_by_position(output_column_id).column = std::move(dst); } }; do_lazy_materialize(_right_output_slot_flags, _build_indexs, (int)_right_col_idx, _build_block.get(), false); - // probe side indexs must be incremental so set try_all_match_one to true + // Probe-side indices are incremental, so check whether they form a contiguous range. do_lazy_materialize(_left_output_slot_flags, _probe_indexs, 0, &_parent->_probe_block, true); return Status::OK(); } diff --git a/regression-test/data/query_p0/join/test_hash_join_probe_side_zero_copy.out b/regression-test/data/query_p0/join/test_hash_join_probe_side_zero_copy.out new file mode 100644 index 00000000000..80a9128c28a --- /dev/null +++ b/regression-test/data/query_p0/join/test_hash_join_probe_side_zero_copy.out @@ -0,0 +1,16 @@ +-- This file is automatically generated. You should know what you did if you want to edit this +-- !pending_build_match -- +1 10 a 100 +2 20 b 200 +3 30 c 300 +3 30 c 301 + +-- !other_join_conjunct -- +1 10 a 100 + +-- !mark_and_post_join_conjunct -- +2 20 b + +-- !reuse_probe_block_after_transfer -- +4 40 +6 60 diff --git a/regression-test/suites/query_p0/join/test_hash_join_probe_side_zero_copy.groovy b/regression-test/suites/query_p0/join/test_hash_join_probe_side_zero_copy.groovy new file mode 100644 index 00000000000..655387838ee --- /dev/null +++ b/regression-test/suites/query_p0/join/test_hash_join_probe_side_zero_copy.groovy @@ -0,0 +1,111 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +suite("test_hash_join_probe_side_zero_copy", "query,p0") { + sql "DROP TABLE IF EXISTS test_hash_join_probe_side_zero_copy_probe" + sql "DROP TABLE IF EXISTS test_hash_join_probe_side_zero_copy_build_many" + sql "DROP TABLE IF EXISTS test_hash_join_probe_side_zero_copy_build_unique" + + sql """ + CREATE TABLE test_hash_join_probe_side_zero_copy_probe ( + k INT NOT NULL, + v INT NOT NULL, + s VARCHAR(10) NOT NULL, + n INT NULL + ) DISTRIBUTED BY HASH(k) BUCKETS 1 + PROPERTIES("replication_num" = "1") + """ + sql """ + CREATE TABLE test_hash_join_probe_side_zero_copy_build_many ( + k INT NOT NULL, + v INT NOT NULL, + s VARCHAR(10) NOT NULL + ) DISTRIBUTED BY HASH(k) BUCKETS 1 + PROPERTIES("replication_num" = "1") + """ + sql """ + CREATE TABLE test_hash_join_probe_side_zero_copy_build_unique ( + k INT NOT NULL, + v INT NOT NULL, + s VARCHAR(10) NOT NULL + ) DISTRIBUTED BY HASH(k) BUCKETS 1 + PROPERTIES("replication_num" = "1") + """ + + sql """ + INSERT INTO test_hash_join_probe_side_zero_copy_probe VALUES + (1, 10, 'a', 10), (2, 20, 'b', 20), (3, 30, 'c', 30), + (4, 40, 'd', 40), (5, 50, 'e', 50), (6, 60, 'f', 60) + """ + sql """ + INSERT INTO test_hash_join_probe_side_zero_copy_build_many VALUES + (1, 100, 'x'), (2, 200, 'y'), (3, 300, 'z'), (3, 301, 'zz') + """ + sql """ + INSERT INTO test_hash_join_probe_side_zero_copy_build_unique VALUES + (1, 100, 'x'), (2, 15, 'y'), (3, 300, 'c') + """ + + // The first output batch has an identity probe mapping, but probe row 3 still has a pending + // build-side match. Probe columns must not be moved until that pending state is exhausted. + order_qt_pending_build_match """ + SELECT /*+SET_VAR(batch_size=3, disable_join_reorder=true)*/ + p.k, p.v, p.s, b.v + FROM test_hash_join_probe_side_zero_copy_probe p + JOIN [broadcast] test_hash_join_probe_side_zero_copy_build_many b + ON p.k = b.k + ORDER BY p.k, p.v, p.s, b.v + """ + + // Other join conjuncts may filter the identity-mapped output after ownership transfer. The + // lazy probe column k is materialized from the still row-sized probe block after filtering. + order_qt_other_join_conjunct """ + SELECT /*+SET_VAR(batch_size=3, disable_join_reorder=true)*/ + p.k, p.v, p.s, b.v + FROM test_hash_join_probe_side_zero_copy_probe p + JOIN [broadcast] test_hash_join_probe_side_zero_copy_build_unique b + ON p.k = b.k AND p.v < b.v AND p.s != b.s + ORDER BY p.k, p.v, p.s, b.v + """ + + // The IN predicate is evaluated by a mark hash join, while the OR predicate is evaluated + // afterwards. The two probe columns used by other join conjuncts make this path eligible for + // probe-side ownership transfer. + order_qt_mark_and_post_join_conjunct """ + SELECT /*+SET_VAR(batch_size=3, disable_join_reorder=true)*/ + p.k, p.v, p.s + FROM test_hash_join_probe_side_zero_copy_probe p + WHERE p.v IN ( + SELECT b.v + FROM test_hash_join_probe_side_zero_copy_build_unique b + WHERE p.k = b.k + 10 AND p.s != b.s AND p.v < b.v + 1000 + ) OR p.k = 2 + ORDER BY p.k, p.v, p.s + """ + + // The first probe batch has only equality misses, so its one-to-one sentinel rows allow + // ownership transfer. The cleared placeholders must become physical reusable columns before + // the second probe batch, whose duplicate probe indices take the normal indexed-copy path. + order_qt_reuse_probe_block_after_transfer """ + SELECT /*+SET_VAR(batch_size=3, disable_join_reorder=true, parallel_pipeline_task_num=1)*/ + p.k, p.n + FROM test_hash_join_probe_side_zero_copy_probe p + LEFT SEMI JOIN [broadcast] test_hash_join_probe_side_zero_copy_build_unique b + ON p.k = b.k + 3 AND p.n < b.v + ORDER BY p.k, p.n + """ +} --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
