This is an automated email from the ASF dual-hosted git repository.
HappenLee pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new 4809030bd52 [improvement](be) Optimize hash join probe side output
(#65300)
4809030bd52 is described below
commit 4809030bd52ffea587254a3b01746e91f755ee61
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 -->
---
.../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 277da08c01e..03401a089f3 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
@@ -33,24 +33,25 @@
namespace doris {
-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());
}
}
@@ -84,15 +85,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>
@@ -154,12 +155,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]) {
@@ -176,8 +200,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());
}
@@ -465,9 +499,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>
@@ -627,7 +661,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] &&
@@ -645,7 +679,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 =
@@ -657,13 +691,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]
