Coverage Report

Created: 2026-10-23 15:57

next uncovered line (L), next uncovered region (R), next uncovered branch (B)
be/src/util/proto_util.h
Line
Count
Source
1
// Licensed to the Apache Software Foundation (ASF) under one
2
// or more contributor license agreements.  See the NOTICE file
3
// distributed with this work for additional information
4
// regarding copyright ownership.  The ASF licenses this file
5
// to you under the Apache License, Version 2.0 (the
6
// "License"); you may not use this file except in compliance
7
// with the License.  You may obtain a copy of the License at
8
//
9
//   http://www.apache.org/licenses/LICENSE-2.0
10
//
11
// Unless required by applicable law or agreed to in writing,
12
// software distributed under the License is distributed on an
13
// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
14
// KIND, either express or implied.  See the License for the
15
// specific language governing permissions and limitations
16
// under the License.
17
18
#pragma once
19
20
#include <brpc/http_method.h>
21
#include <gen_cpp/internal_service.pb.h>
22
23
#include "common/config.h"
24
#include "common/status.h"
25
#include "runtime/exec_env.h"
26
#include "runtime/runtime_state.h"
27
#include "util/brpc_client_cache.h"
28
#include "util/network_util.h"
29
30
namespace doris {
31
32
// When the tuple/block data is greater than 2G, embed the tuple/block data
33
// and the request serialization string in the attachment, and use "http" brpc.
34
// "http"brpc requires that only one of request and attachment be non-null.
35
//
36
// 2G: In the default "baidu_std" brpcd, upper limit of the request and attachment length is 2G.
37
constexpr size_t MIN_HTTP_BRPC_SIZE = (1ULL << 31);
38
39
// Embed column_values and brpc request serialization string in controller attachment.
40
template <typename Params, typename Closure>
41
Status request_embed_attachment_contain_blockv2(Params* brpc_request,
42
0
                                                std::unique_ptr<Closure>& closure) {
43
0
    std::string column_values = std::move(*brpc_request->mutable_block()->mutable_column_values());
44
0
    brpc_request->mutable_block()->mutable_column_values()->clear();
45
0
    return request_embed_attachmentv2(brpc_request, column_values, closure);
46
0
}
Unexecuted instantiation: _ZN5doris40request_embed_attachment_contain_blockv2INS_28PTabletWriterAddBlockRequestENS_18AutoReleaseClosureIS1_NS_18WriteBlockCallbackINS_27PTabletWriterAddBlockResultEEEEEEENS_6StatusEPT_RSt10unique_ptrIT0_St14default_deleteISB_EE
Unexecuted instantiation: _ZN5doris40request_embed_attachment_contain_blockv2INS_19PTransmitDataParamsENS_18AutoReleaseClosureIS1_NS_20ExchangeSendCallbackINS_19PTransmitDataResultEEEEEEENS_6StatusEPT_RSt10unique_ptrIT0_St14default_deleteISB_EE
47
48
388k
inline bool enable_http_send_block(const PTransmitDataParams& request) {
49
388k
    if (!config::transfer_large_data_by_brpc) {
50
0
        return false;
51
0
    }
52
388k
    if (!request.has_block() || !request.block().has_column_values()) {
53
384k
        return false;
54
384k
    }
55
4.30k
    if (request.ByteSizeLong() < MIN_HTTP_BRPC_SIZE) {
56
2
        return false;
57
2
    }
58
4.30k
    return true;
59
4.30k
}
60
61
template <typename Closure>
62
375k
void transmit_blockv2(PBackendService_Stub* stub, std::unique_ptr<Closure> closure) {
63
375k
    closure->cntl_->http_request().Clear();
64
375k
    stub->transmit_block(closure->cntl_.get(), closure->request_.get(), closure->response_.get(),
65
375k
                         closure.get());
66
375k
    closure.release();
67
375k
}
68
69
template <typename Closure>
70
10.4k
void transmit_serialized_block(PBackendService_Stub* stub, std::unique_ptr<Closure> closure) {
71
    // SerializedRequest bypasses brpc's protobuf serializer, including request compression.
72
    // Exchange compresses column_values before preparing the shared protobuf fragment.
73
10.4k
    DCHECK_EQ(closure->cntl_->request_compress_type(), brpc::COMPRESS_TYPE_NONE);
74
10.4k
    static const auto* method = PBackendService::descriptor()->FindMethodByName("transmit_block");
75
10.4k
    closure->cntl_->http_request().Clear();
76
10.4k
    stub->channel()->CallMethod(method, closure->cntl_.get(), closure->request_.get(),
77
10.4k
                                closure->response_.get(), closure.get());
78
10.4k
    closure.release();
79
10.4k
}
80
81
template <typename Closure>
82
Status transmit_block_httpv2(ExecEnv* exec_env, std::unique_ptr<Closure> closure,
83
0
                             TNetworkAddress brpc_dest_addr) {
84
0
    RETURN_IF_ERROR(request_embed_attachment_contain_blockv2(closure->request_.get(), closure));
85
86
0
    std::string host = brpc_dest_addr.hostname;
87
0
    auto dns_cache = ExecEnv::GetInstance()->dns_cache();
88
0
    if (dns_cache == nullptr) {
89
0
        LOG(WARNING) << "DNS cache is not initialized, skipping hostname resolve";
90
0
    } else if (!is_valid_ip(brpc_dest_addr.hostname)) {
91
0
        Status status = dns_cache->get(brpc_dest_addr.hostname, &host);
92
0
        if (!status.ok()) {
93
0
            LOG(WARNING) << "failed to get ip from host " << brpc_dest_addr.hostname << ": "
94
0
                         << status.to_string();
95
0
            return Status::InternalError("failed to get ip from host {}", brpc_dest_addr.hostname);
96
0
        }
97
0
    }
98
    //format an ipv6 address
99
0
    std::string brpc_url = get_brpc_http_url(brpc_dest_addr.hostname, brpc_dest_addr.port);
100
101
0
    std::shared_ptr<PBackendService_Stub> brpc_http_stub =
102
0
            exec_env->brpc_internal_client_cache()->get_new_client_no_cache(
103
0
                    brpc_url, "http", "", "", 2000, 10, brpc_dest_addr.hostname);
104
0
    if (brpc_http_stub == nullptr) {
105
0
        return Status::InternalError("failed to open brpc http client to {}", brpc_url);
106
0
    }
107
0
    closure->cntl_->http_request().uri() =
108
0
            brpc_url + "/PInternalServiceImpl/transmit_block_by_http";
109
0
    closure->cntl_->http_request().set_method(brpc::HTTP_METHOD_POST);
110
0
    closure->cntl_->http_request().set_content_type("application/json");
111
0
    brpc_http_stub->transmit_block_by_http(closure->cntl_.get(), nullptr, closure->response_.get(),
112
0
                                           closure.get());
113
0
    closure.release();
114
115
0
    return Status::OK();
116
0
}
117
118
template <typename Params, typename Closure>
119
Status request_embed_attachmentv2(Params* brpc_request, const std::string& data,
120
0
                                  std::unique_ptr<Closure>& closure) {
121
0
    butil::IOBuf attachment;
122
123
    // step1: serialize brpc_request to string, and append to attachment.
124
0
    std::string req_str;
125
0
    if (!brpc_request->SerializeToString(&req_str)) {
126
0
        return Status::InternalError("failed to serialize the request");
127
0
    }
128
0
    int64_t req_str_size = req_str.size();
129
0
    attachment.append(&req_str_size, sizeof(req_str_size));
130
0
    attachment.append(req_str);
131
132
    // step2: append data to attachment and put it in the closure.
133
0
    int64_t data_size = data.size();
134
0
    attachment.append(&data_size, sizeof(data_size));
135
0
    try {
136
0
        attachment.append(data);
137
0
    } catch (...) {
138
0
        LOG(WARNING) << "Try to alloc " << data_size
139
0
                     << " bytes for append data to attachment failed. ";
140
0
        return Status::MemoryAllocFailed("request embed attachment failed to memcpy {} bytes",
141
0
                                         data_size);
142
0
    }
143
    // step3: attachment add to closure.
144
0
    closure->cntl_->request_attachment().swap(attachment);
145
0
    return Status::OK();
146
0
}
Unexecuted instantiation: _ZN5doris26request_embed_attachmentv2INS_28PTabletWriterAddBlockRequestENS_18AutoReleaseClosureIS1_NS_18WriteBlockCallbackINS_27PTabletWriterAddBlockResultEEEEEEENS_6StatusEPT_RKNSt7__cxx1112basic_stringIcSt11char_traitsIcESaIcEEERSt10unique_ptrIT0_St14default_deleteISJ_EE
Unexecuted instantiation: _ZN5doris26request_embed_attachmentv2INS_19PTransmitDataParamsENS_18AutoReleaseClosureIS1_NS_20ExchangeSendCallbackINS_19PTransmitDataResultEEEEEEENS_6StatusEPT_RKNSt7__cxx1112basic_stringIcSt11char_traitsIcESaIcEEERSt10unique_ptrIT0_St14default_deleteISJ_EE
147
148
// Extract the brpc request and block from the controller attachment,
149
// and put the block into the request.
150
template <typename Params>
151
0
Status attachment_extract_request_contain_block(Params* brpc_request, brpc::Controller* cntl) {
152
0
    auto block = brpc_request->mutable_block();
153
0
    return attachment_extract_request(brpc_request, cntl, block->mutable_column_values());
154
0
}
Unexecuted instantiation: _ZN5doris40attachment_extract_request_contain_blockINS_28PTabletWriterAddBlockRequestEEENS_6StatusEPT_PN4brpc10ControllerE
Unexecuted instantiation: _ZN5doris40attachment_extract_request_contain_blockINS_19PTransmitDataParamsEEENS_6StatusEPT_PN4brpc10ControllerE
155
156
template <typename Params>
157
0
Status attachment_extract_request(Params* brpc_request, brpc::Controller* cntl, std::string* data) {
158
0
    const butil::IOBuf& io_buf = cntl->request_attachment();
159
160
    // step1: deserialize request string to brpc_request from attachment.
161
0
    int64_t req_str_size;
162
0
    io_buf.copy_to(&req_str_size, sizeof(req_str_size), 0);
163
0
    std::string req_str;
164
0
    io_buf.copy_to(&req_str, req_str_size, sizeof(req_str_size));
165
0
    brpc_request->ParseFromString(req_str);
166
167
    // step2: extract data from attachment.
168
0
    int64_t data_size;
169
0
    io_buf.copy_to(&data_size, sizeof(data_size), sizeof(req_str_size) + req_str_size);
170
0
    try {
171
0
        io_buf.copy_to(data, data_size, sizeof(data_size) + sizeof(req_str_size) + req_str_size);
172
0
    } catch (...) {
173
0
        LOG(WARNING) << "Try to alloc " << data_size
174
0
                     << " bytes for extract data from attachment failed. ";
175
0
        return Status::MemoryAllocFailed("attachment extract request failed to memcpy {} bytes",
176
0
                                         data_size);
177
0
    }
178
0
    return Status::OK();
179
0
}
Unexecuted instantiation: _ZN5doris26attachment_extract_requestINS_28PTabletWriterAddBlockRequestEEENS_6StatusEPT_PN4brpc10ControllerEPNSt7__cxx1112basic_stringIcSt11char_traitsIcESaIcEEE
Unexecuted instantiation: _ZN5doris26attachment_extract_requestINS_19PTransmitDataParamsEEENS_6StatusEPT_PN4brpc10ControllerEPNSt7__cxx1112basic_stringIcSt11char_traitsIcESaIcEEE
180
181
} // namespace doris