This is an automated email from the ASF dual-hosted git repository.
yiguolei pushed a commit to branch branch-4.1
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/branch-4.1 by this push:
new 341c927a1c9 branch-4.1: [improvement](be) Optimize hash join probe
side output (#65600)
341c927a1c9 is described below
commit 341c927a1c9e5cc190f0ce1886becf8186e7d4d5
Author: Pxl <[email protected]>
AuthorDate: Wed Jul 15 12:07:10 2026 +0800
branch-4.1: [improvement](be) Optimize hash join probe side output (#65600)
### What problem does this PR solve?
Issue Number: None
Related PR: #65300
Problem Summary: Backport #65300 to branch-4.1. Hash join probe-side
output can copy projected probe columns even when the output rows map
one-to-one to the consumed probe block. This change transfers eligible
probe-side columns into the output block and avoids the redundant copy
for supported join types, while retaining conservative guards for
filtering, nullable-output, and variant-finalization cases.
The original regression coverage `test_hash_join_probe_side_zero_copy`
is included in this backport.
### Release note
None
### Check List (For Author)
- Test:
- `inner_join_impl.cpp` and `asof_left_outer_join_impl.cpp` passed
`-fsyntax-only` with the branch-4.1 release compile commands
- `clang-format --dry-run --Werror` on the changed hash-join headers
- `git diff --check upstream/branch-4.1..HEAD`
- Behavior changed: No
- Does this need documentation: No
---
.../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]
