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 762147325b7 branch-4.1: [fix](topn) Handle empty rowid fetch RPC
failures #66443 (#66522)
762147325b7 is described below
commit 762147325b773f982737045ad4bffc403dbf036a
Author: github-actions[bot]
<41898282+github-actions[bot]@users.noreply.github.com>
AuthorDate: Fri Aug 7 14:25:56 2026 +0800
branch-4.1: [fix](topn) Handle empty rowid fetch RPC failures #66443
(#66522)
Cherry-picked from #66443
Co-authored-by: HappenLee <[email protected]>
---
be/src/exec/operator/materialization_opertor.cpp | 57 ++++++++++++++--------
be/src/exec/operator/materialization_opertor.h | 1 +
.../operator/materialization_shared_state_test.cpp | 41 ++++++++++++++--
3 files changed, 75 insertions(+), 24 deletions(-)
diff --git a/be/src/exec/operator/materialization_opertor.cpp
b/be/src/exec/operator/materialization_opertor.cpp
index 5966f46b601..629df2ad52b 100644
--- a/be/src/exec/operator/materialization_opertor.cpp
+++ b/be/src/exec/operator/materialization_opertor.cpp
@@ -279,6 +279,12 @@ Status
MaterializationSharedState::merge_multi_response(RuntimeProfile* profile)
// Phase 1: Deserialize the i-th response block from every BE into
block_maps.
// Each BE's response.blocks(i) corresponds to the i-th relation's
fetched columns.
for (auto& [backend_id, rpc_struct] : rpc_struct_map) {
+ const auto request_row_count =
rpc_struct.request.request_block_descs(i).row_id_size();
+ // An empty request may have no response block when its RPC fails.
It cannot
+ // contribute rows to block_order_results, so there is nothing to
deserialize.
+ if (request_row_count == 0) {
+ continue;
+ }
Block partial_block;
size_t uncompressed_size = 0;
int64_t uncompressed_time = 0;
@@ -292,13 +298,12 @@ Status
MaterializationSharedState::merge_multi_response(RuntimeProfile* profile)
// refer 'if (!id_file_map)' in RowIdStorageReader::read_by_rowids.
// 2. Report error in any case where the row count doesn't match,
even if it's not empty,
// since that indicates a bug in BE's row fetching logic or
serialization logic.
- if (rpc_struct.request.request_block_descs(i).row_id_size() !=
partial_block.rows()) {
+ if (request_row_count != partial_block.rows()) {
return Status::InternalError(
fmt::format("merge_multi_response, "
"backend_id {} returned block with row
count {} not match "
"request row id count {}",
- backend_id, partial_block.rows(),
-
rpc_struct.request.request_block_descs(i).row_id_size()));
+ backend_id, partial_block.rows(),
request_row_count));
}
if (rpc_struct.response.blocks(i).has_profile()) {
auto response_profile =
@@ -306,8 +311,7 @@ Status
MaterializationSharedState::merge_multi_response(RuntimeProfile* profile)
_update_profile_info(backend_id, response_profile.get());
}
- // Only insert non-empty blocks. A BE may return an empty block if
- // request.request_block_descs(i).row_id_size() is 0
+ // Only insert non-empty blocks.
if (!partial_block.is_empty_column()) {
// Reset row cursor to 0 — we'll consume rows from this block
sequentially.
block_maps[backend_id] =
std::make_pair(std::move(partial_block), 0);
@@ -368,6 +372,29 @@ Status
MaterializationSharedState::merge_multi_response(RuntimeProfile* profile)
return Status::OK();
}
+Status MaterializationSharedState::validate_rpc_results(int node_id) {
+ for (auto& [backend_id, rpc_struct] : rpc_struct_map) {
+ if (rpc_struct.cntl->Failed()) {
+ if (count_request_rows(rpc_struct.request) > 0) {
+ return Status::InternalError(
+ "Failed to send brpc request, error_text=" +
rpc_struct.cntl->ErrorText() +
+ " Materialization Sink node id:" +
std::to_string(node_id) +
+ " target_backend_id:" + std::to_string(backend_id));
+ }
+ rpc_struct.cntl->Reset();
+ continue;
+ }
+ if (rpc_struct.response.status().status_code() != 0) {
+ Status st = Status::create(rpc_struct.response.status());
+ st.append(fmt::format(", Backend:{}, Materialization Sink node
id:{}", backend_id,
+ node_id));
+ return st;
+ }
+ rpc_struct.cntl->Reset();
+ }
+ return Status::OK();
+}
+
void MaterializationSharedState::_update_profile_info(int64_t backend_id,
RuntimeProfile*
response_profile) {
if (!backend_profile_info_string.contains(backend_id)) {
@@ -608,6 +635,9 @@ Status MaterializationOperator::push(RuntimeState* state,
Block* in_block, bool
MonotonicStopWatch rpc_timer(true);
for (auto& [backend_id, rpc_struct] :
local_state._materialization_state.rpc_struct_map) {
auto* callback = brpc::NewCallback(fetch_callback, &counter);
+ // The response object is reused across batches. Clear it so an
ignored failure for
+ // an empty request cannot expose a response left by an earlier
batch.
+ rpc_struct.response.Clear();
rpc_struct.cntl->set_timeout_ms(state->execution_timeout() * 1000);
// send brpc request
rpc_struct.stub->multiget_data_v2(rpc_struct.cntl.get(),
&rpc_struct.request,
@@ -618,22 +648,7 @@ Status MaterializationOperator::push(RuntimeState* state,
Block* in_block, bool
local_state._max_rpc_timer->set(static_cast<int64_t>(time));
}
- for (auto& [backend_id, rpc_struct] :
local_state._materialization_state.rpc_struct_map) {
- if (rpc_struct.cntl->Failed()) {
- std::string error_text =
- "Failed to send brpc request, error_text=" +
rpc_struct.cntl->ErrorText() +
- " Materialization Sink node id:" +
std::to_string(node_id()) +
- " target_backend_id:" + std::to_string(backend_id);
- return Status::InternalError(error_text);
- }
- if (rpc_struct.response.status().status_code() != 0) {
- Status st = Status::create(rpc_struct.response.status());
- st.append(fmt::format(", Backend:{}, Materialization Sink node
id:{}", backend_id,
- node_id()));
- return st;
- }
- rpc_struct.cntl->Reset();
- }
+
RETURN_IF_ERROR(local_state._materialization_state.validate_rpc_results(node_id()));
if (local_state._materialization_state.need_merge_block) {
SCOPED_TIMER(local_state._merge_response_timer);
diff --git a/be/src/exec/operator/materialization_opertor.h
b/be/src/exec/operator/materialization_opertor.h
index c32fbb194ca..3a180152a01 100644
--- a/be/src/exec/operator/materialization_opertor.h
+++ b/be/src/exec/operator/materialization_opertor.h
@@ -47,6 +47,7 @@ public:
Status init_multi_requests(const TMaterializationNode& tnode,
RuntimeState* state);
Status create_muiltget_result(const Columns& columns, bool eos, bool
gc_id_map);
+ Status validate_rpc_results(int node_id);
Status merge_multi_response(RuntimeProfile* profile);
void get_block(Block* block);
diff --git a/be/test/exec/operator/materialization_shared_state_test.cpp
b/be/test/exec/operator/materialization_shared_state_test.cpp
index d6cfca699ba..3632e479a23 100644
--- a/be/test/exec/operator/materialization_shared_state_test.cpp
+++ b/be/test/exec/operator/materialization_shared_state_test.cpp
@@ -60,6 +60,8 @@ protected:
_shared_state->rpc_struct_map[_backend_id2] = FetchRpcStruct();
_shared_state->rpc_struct_map[_backend_id1].request.add_request_block_descs();
_shared_state->rpc_struct_map[_backend_id2].request.add_request_block_descs();
+ _shared_state->rpc_struct_map[_backend_id1].cntl =
std::make_unique<brpc::Controller>();
+ _shared_state->rpc_struct_map[_backend_id2].cntl =
std::make_unique<brpc::Controller>();
}
std::shared_ptr<MaterializationSharedState> _shared_state;
@@ -69,6 +71,39 @@ protected:
int64_t _backend_id2;
};
+TEST_F(MaterializationSharedStateTest,
TestRpcFailureWithoutFetchRowsIsIgnored) {
+ auto& rpc_struct = _shared_state->rpc_struct_map[_backend_id1];
+ rpc_struct.cntl->SetFailed("injected connection failure");
+
+ Status st = _shared_state->validate_rpc_results(10);
+
+ EXPECT_TRUE(st.ok()) << st.to_string();
+ EXPECT_FALSE(rpc_struct.cntl->Failed());
+}
+
+TEST_F(MaterializationSharedStateTest, TestRpcFailureWithFetchRowsFails) {
+ auto& rpc_struct = _shared_state->rpc_struct_map[_backend_id1];
+ add_request_row(rpc_struct.request.mutable_request_block_descs(0), 0, 1);
+ rpc_struct.cntl->SetFailed("injected connection failure");
+
+ Status st = _shared_state->validate_rpc_results(10);
+
+ EXPECT_FALSE(st.ok());
+ EXPECT_NE(st.to_string().find("target_backend_id:1001"),
std::string::npos);
+}
+
+TEST_F(MaterializationSharedStateTest, TestNonOkResponseWithoutFetchRowsFails)
{
+ auto& rpc_struct = _shared_state->rpc_struct_map[_backend_id1];
+ Status::InternalError("injected remote failure")
+ .to_protobuf(rpc_struct.response.mutable_status());
+
+ Status st = _shared_state->validate_rpc_results(10);
+
+ EXPECT_FALSE(st.ok());
+ EXPECT_NE(st.to_string().find("injected remote failure"),
std::string::npos);
+ EXPECT_NE(st.to_string().find("Backend:1001"), std::string::npos);
+}
+
TEST_F(MaterializationSharedStateTest, TestCreateMultiGetResult) {
// Create test columns for rowids
Columns columns;
@@ -453,7 +488,7 @@ TEST_F(MaterializationSharedStateTest,
TestMergeMultiResponseStaleBlockMaps) {
_shared_state->rpc_struct_map[_backend_id1].request.add_request_block_descs();
_shared_state->rpc_struct_map[_backend_id2].request.add_request_block_descs();
- // --- Build BE_1's response: blocks[0]=1 row (INT), blocks[1]=empty ---
+ // --- Build BE_1's response: blocks[0]=1 row (INT), no block for relation
1 ---
{
add_request_row(
_shared_state->rpc_struct_map[_backend_id1].request.mutable_request_block_descs(0),
@@ -473,8 +508,8 @@ TEST_F(MaterializationSharedStateTest,
TestMergeMultiResponseStaleBlockMaps) {
ASSERT_TRUE(rel0_block.serialize(0, pb0, &us, &cs, &ct,
CompressionTypePB::LZ4).ok());
_shared_state->response_blocks[0] = rel0_block.clone_empty();
- // blocks[1]: empty (BE_1 has no data for relation 1)
- response.add_blocks();
+ // BE_1 has no data for relation 1. This also models an ignored
transport failure for
+ // an empty RPC: there is no response block to deserialize for that
relation.
_shared_state->rpc_struct_map[_backend_id1].response =
std::move(response);
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
