This is an automated email from the ASF dual-hosted git repository.
gabriellee 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 5d7b7c560bc [fix](coordinator) Fix wrong `recvrId` in fragment
contains BHJ (#47727)
5d7b7c560bc is described below
commit 5d7b7c560bcdcd27a14c97540375bf2fbfc44a12
Author: Gabriel <liwenqi...@selectdb.com>
AuthorDate: Tue Feb 11 10:44:15 2025 +0800
[fix](coordinator) Fix wrong `recvrId` in fragment contains BHJ (#47727)
In Coordinator, a shuffle map consists of `recvrId` in each instance.
For example, if 3 BEs exist in a cluster, for a shuffled hash join, we
get 3 maps for a fragment sent to each BE:
BE0: {0:0, 1:1}
BE1: {2:0, 3:1}
BE2: {4:0, 5:1}
In this example, parallelism is 2. Keys in shuffle map indicate the
global shuffle id and the values indicate the instance id in current BE.
In Coordinator, the `recvrId` is the global shuffle id of each instance
so we may get a wrong result if it is wrong.
This bug is caused by `recvrId` set by a BHJ fragment. If a fragment
contains both BHJ and SHJ, `recvrId` should be set by SHJ and BHJ should
be ignored.
---
be/src/pipeline/local_exchange/local_exchanger.cpp | 14 ++++++++++++++
.../src/main/java/org/apache/doris/qe/Coordinator.java | 2 --
2 files changed, 14 insertions(+), 2 deletions(-)
diff --git a/be/src/pipeline/local_exchange/local_exchanger.cpp
b/be/src/pipeline/local_exchange/local_exchanger.cpp
index c011171dbe3..c768668acc4 100644
--- a/be/src/pipeline/local_exchange/local_exchanger.cpp
+++ b/be/src/pipeline/local_exchange/local_exchanger.cpp
@@ -214,16 +214,30 @@ Status ShuffleExchanger::_split_rows(RuntimeState* state,
const uint32_t* __rest
*/
DCHECK(shuffle_idx_to_instance_idx && shuffle_idx_to_instance_idx->size()
> 0);
const auto& map = *shuffle_idx_to_instance_idx;
+ int32_t enqueue_rows = 0;
for (const auto& it : map) {
DCHECK(it.second >= 0 && it.second < _num_partitions)
<< it.first << " : " << it.second << " " << _num_partitions;
uint32_t start = partition_rows_histogram[it.first];
uint32_t size = partition_rows_histogram[it.first + 1] - start;
if (size > 0) {
+ enqueue_rows += size;
_enqueue_data_and_set_ready(it.second, local_state,
{new_block_wrapper, {row_idx, start,
size}});
}
}
+ if (enqueue_rows != rows) [[unlikely]] {
+ fmt::memory_buffer debug_string_buffer;
+ fmt::format_to(debug_string_buffer, "Type: {}, Local Exchange Id: {},
Shuffled Map: ",
+ get_exchange_type_name(get_type()),
local_state->parent()->node_id());
+ for (const auto& it : map) {
+ fmt::format_to(debug_string_buffer, "[{}:{}], ", it.first,
it.second);
+ }
+ return Status::InternalError(
+ "Rows mismatched! Data may be lost. [Expected enqueue rows={},
Real enqueue "
+ "rows={}, Detail: {}]",
+ rows, enqueue_rows, fmt::to_string(debug_string_buffer));
+ }
return Status::OK();
}
diff --git a/fe/fe-core/src/main/java/org/apache/doris/qe/Coordinator.java
b/fe/fe-core/src/main/java/org/apache/doris/qe/Coordinator.java
index 3cc60940b05..fcb549736ef 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/qe/Coordinator.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/qe/Coordinator.java
@@ -1443,7 +1443,6 @@ public class Coordinator implements CoordInterface {
} else {
destHosts.put(param.host, param);
TPlanFragmentDestination dest = new
TPlanFragmentDestination();
- param.recvrId = params.destinations.size();
dest.fragment_instance_id = param.instanceId;
try {
dest.server = toRpcHost(param.host);
@@ -1589,7 +1588,6 @@ public class Coordinator implements CoordInterface {
destHosts.put(param.host, param);
TPlanFragmentDestination dest = new
TPlanFragmentDestination();
dest.fragment_instance_id = param.instanceId;
- param.recvrId = params.destinations.size();
try {
dest.server = toRpcHost(param.host);
dest.setBrpcServer(toBrpcHost(param.host));
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscr...@doris.apache.org
For additional commands, e-mail: commits-h...@doris.apache.org
