github-actions[bot] commented on code in PR #63258:
URL: https://github.com/apache/doris/pull/63258#discussion_r4286809662
##########
be/src/util/proto_util.h:
##########
@@ -36,26 +42,131 @@ namespace doris {
// 2G: In the default "baidu_std" brpcd, upper limit of the request and
attachment length is 2G.
constexpr size_t MIN_HTTP_BRPC_SIZE = (1ULL << 31);
-// Embed column_values and brpc request serialization string in controller
attachment.
+// Attachment layout shared by every http send path:
+//
+// [int64 request_size][serialized request][int64 payload_size][payload]...
+//
+// The serialized request carries everything but the block payloads
(PBlock::column_values).
+// One payload section follows per payload block of the request, in block
order, see
+// payload_blocks(). A request with a single block therefore keeps the layout
older
+// receivers expect; receivers older than this layout read exactly one payload
into the
+// singular `block`, so a multi-payload attachment (repeated `blocks`) is only
understood by
+// receivers that carry it. Such requests could not be sent at all before
(plain brpc rejected
+// them for exceeding its size limit), so no working mixed-version case is
lost.
+
+// The blocks whose column_values travel as attachment payloads. A generic
request carries one
+// block; PTransmitDataParams carries either the singular `block` or the
repeated `blocks`.
+template <typename Params>
+std::vector<PBlock*> payload_blocks(Params* brpc_request) {
+ return {brpc_request->mutable_block()};
+}
+
+inline std::vector<PBlock*> payload_blocks(PTransmitDataParams* brpc_request) {
+ if (brpc_request->has_block()) {
+ DCHECK(brpc_request->blocks().empty());
+ return {brpc_request->mutable_block()};
+ }
+ std::vector<PBlock*> blocks;
+ blocks.reserve(brpc_request->blocks_size());
+ for (auto& block : *brpc_request->mutable_blocks()) {
+ blocks.push_back(&block);
+ }
+ return blocks;
+}
+
+// Append one `[int64 size][bytes]` section to the attachment.
+inline Status attachment_append_section(butil::IOBuf* attachment, const
std::string& data) {
+ int64_t data_size = data.size();
+ attachment->append(&data_size, sizeof(data_size));
+ try {
+ attachment->append(data);
+ } catch (...) {
+ LOG(WARNING) << "Try to alloc " << data_size
+ << " bytes for append data to attachment failed. ";
+ return Status::MemoryAllocFailed("request embed attachment failed to
memcpy {} bytes",
+ data_size);
+ }
+ return Status::OK();
+}
+
+// Embed `brpc_request` followed by `payloads`, one section each, in the
controller attachment.
+template <typename Params, typename Closure>
+Status request_embed_attachmentv2(Params* brpc_request,
+ const std::vector<const std::string*>&
payloads,
+ std::unique_ptr<Closure>& closure) {
+ std::string req_str;
+ if (!brpc_request->SerializeToString(&req_str)) {
+ return Status::InternalError("failed to serialize the request");
+ }
+ butil::IOBuf attachment;
+ RETURN_IF_ERROR(attachment_append_section(&attachment, req_str));
+ for (const auto* payload : payloads) {
+ RETURN_IF_ERROR(attachment_append_section(&attachment, *payload));
+ }
+ closure->cntl_->request_attachment().swap(attachment);
+ return Status::OK();
+}
+
+template <typename Params, typename Closure>
+Status request_embed_attachmentv2(Params* brpc_request, const std::string&
data,
+ std::unique_ptr<Closure>& closure) {
+ return request_embed_attachmentv2(brpc_request, std::vector<const
std::string*> {&data},
+ closure);
+}
+
+// Embed a request that owns its block: column_values is moved into the
attachment and the
+// request keeps the field present but empty.
template <typename Params, typename Closure>
Status request_embed_attachment_contain_blockv2(Params* brpc_request,
std::unique_ptr<Closure>&
closure) {
- std::string column_values =
std::move(*brpc_request->mutable_block()->mutable_column_values());
- brpc_request->mutable_block()->mutable_column_values()->clear();
+ auto* block = brpc_request->mutable_block();
+ std::string column_values = std::move(*block->mutable_column_values());
+ block->mutable_column_values()->clear();
return request_embed_attachmentv2(brpc_request, column_values, closure);
}
+// Embed a request whose blocks only borrow their column_values (see
+// ExchangeSinkBuffer / borrow_block()). Each payload is detached from the
request while the
+// request is serialized, appended by reference and attached back, so the
owner of the payload
+// is never read through the request's presence bits nor mutated: the same
owner may be
+// serialized by other channels at the same time.
+template <typename Closure>
+Status request_embed_attachment_borrowed_blocksv2(PTransmitDataParams*
brpc_request,
+ std::unique_ptr<Closure>&
closure) {
+ std::vector<PBlock*> blocks = payload_blocks(brpc_request);
+ std::vector<std::string*> payloads;
+ payloads.reserve(blocks.size());
+ // Registered before any payload is detached: if a placeholder allocation
below throws, the
+ // payloads detached so far are attached back and the placeholders already
installed are freed
+ // by set_allocated_column_values().
+ Defer attach_back([&]() {
+ for (size_t i = 0; i < payloads.size(); ++i) {
+ blocks[i]->set_allocated_column_values(payloads[i]);
+ }
+ });
+ for (auto* block : blocks) {
+ payloads.push_back(block->release_column_values());
+ DCHECK(payloads.back() != nullptr);
+ // Keep the field present on the wire: receivers that take the payload
slot before
+ // parsing the request rely on the serialized field to restore its
presence bit.
+ block->set_column_values("");
+ }
+ return request_embed_attachmentv2(
+ brpc_request, std::vector<const std::string*>(payloads.begin(),
payloads.end()),
+ closure);
+}
+
inline bool enable_http_send_block(const PTransmitDataParams& request) {
if (!config::transfer_large_data_by_brpc) {
return false;
}
- if (!request.has_block() || !request.block().has_column_values()) {
- return false;
- }
- if (request.ByteSizeLong() < MIN_HTTP_BRPC_SIZE) {
+ if (!request.has_block() && request.blocks().empty()) {
Review Comment:
[P1] Keep repeated-block HTTP compatible with older receivers
During a rolling upgrade, a new sender with multi-block exchange enabled can
send a >=2 GiB request through this branch to a BE still running the base
version. That receiver's `attachment_extract_request_contain_block()` only
fills the singular `block.column_values`; it parses the repeated `blocks` but
leaves their payloads empty. `VDataStreamMgr::transmit_block()` then selects
those repeated blocks and enqueues them. With the default
`fragment_transmission_compression_codec = 'none'`, `Block::deserialize()`
passes these empty buffers to unchecked column header decoding
(`deserialize_const_flag_and_row_num`), causing invalid reads/decoding instead
of the former transport rejection. Preserve the legacy singular framing,
negotiate receiver support, or fail before emitting the new shape. This differs
from the fixed singular presence-bit thread: setting the empty repeated fields
present does not teach the old receiver to attach their data.
##########
be/src/exec/operator/exchange_sink_buffer.cpp:
##########
@@ -477,35 +490,14 @@ Status ExchangeSinkBuffer::_send_rpc(RpcInstance&
instance_data) {
return !req.block_holder->serialized_block().empty() ||
req.block_holder->get_block()->column_metas().empty();
});
- auto& request = requests[0];
auto& brpc_request = instance_data.request;
brpc_request->set_sender_id(channel->_parent->sender_id());
brpc_request->set_be_number(channel->_parent->be_number());
- if (_send_multi_blocks && !use_shared_protobuf) {
- for (const auto& req : requests) {
- if (auto block = req.block_holder->get_block();
- block && !block->column_metas().empty()) {
- add_borrowed_broadcast_block(*brpc_request, *block);
- }
- }
- } else if (!_send_multi_blocks) {
- if (request.block_holder->get_block() &&
- !request.block_holder->get_block()->column_metas().empty()) {
-
brpc_request->set_allocated_block(request.block_holder->get_block());
- }
- }
- auto detach_blocks = [&]() {
- if (!_send_multi_blocks && request.block_holder->get_block()) {
- static_cast<void>(brpc_request->release_block());
- } else {
- for (int i = 0; i < brpc_request->mutable_blocks()->size();
++i) {
-
static_cast<void>(brpc_request->mutable_blocks(i)->release_column_values());
- }
- brpc_request->clear_blocks();
- }
- };
- Defer release_block(detach_blocks);
+ Defer release_blocks([&]() {
release_borrowed_blocks(brpc_request.get()); });
+ borrow_blocks(
Review Comment:
[P2] Skip rebuilding blocks for cached broadcast sends
When `_send_multi_blocks` and `use_shared_protobuf` are true, this new
unconditional call deep-copies every block's recursive `column_metas` into a
request for each destination. For normal requests below 2 GiB, the cached
branch immediately clears these blocks and sends the pre-serialized holder
fragments instead. The previous branch skipped construction in this case. A
wide/nested-schema broadcast with many remote destinations and small batches
therefore adds a full schema-copy pass per destination under the RPC mutex for
metadata that is never sent. Compute the HTTP decision from the routing header
and cached fragment sizes, and construct borrowed blocks only for HTTP or the
uncached fallback. This is separate from the earlier metadata-ownership
simplification: no request uses these copies.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
