github-actions[bot] commented on code in PR #68366:
URL: https://github.com/apache/doris/pull/68366#discussion_r4089363894
##########
be/src/load/channel/load_stream.cpp:
##########
@@ -772,6 +838,7 @@ void LoadStream::_dispatch(StreamId id, const
PStreamHeader& hdr, butil::IOBuf*
}
switch (hdr.opcode()) {
+ case PStreamHeader::ADD_PARTIAL_ROWSET:
Review Comment:
[P1] Account for this new header before parsing it. `ADD_PARTIAL_ROWSET`
puts the complete partial `RowsetMetaPB` and MOW delete-bitmap result in
`PStreamHeader`, but `on_received_messages` parses that header before
`_dispatch` attaches `_resource_ctx`. Concurrent writers can therefore
materialize large metadata/bitmaps outside the load/query limiter, then
retain/copy them again inside the attached path, so a valid load can exhaust BE
memory instead of being limited. Attach the target resource context before
parsing, reserve/validate the declared header size, and bound or chunk this
payload.
##########
be/src/load/delta_writer/delta_writer_v2.cpp:
##########
@@ -208,12 +296,51 @@ Status DeltaWriterV2::close_wait(int32_t& num_segments,
RuntimeProfile* profile)
_update_profile(profile);
}
RETURN_IF_ERROR(_memtable_writer->close_wait(profile));
- num_segments = _rowset_writer->next_segment_id();
+ num_segments = _rowset_writer->get_allocated_segment_id() -
_segment_start_id;
+ if (_req.cloud_sink_upload) {
+ RETURN_IF_ERROR(_finish_sink_upload(profile));
+ }
_delta_written_success = true;
return Status::OK();
}
+Status DeltaWriterV2::_finish_sink_upload(RuntimeProfile* profile) {
+ RowsetSharedPtr partial;
+ RETURN_IF_ERROR(_rowset_writer->build(partial));
+ if (UnixSeconds() >= _req.txn_expiration) {
Review Comment:
[P2] Preserve cache-cleanup ownership across cloud load failures. At this
return, `build(partial)` has already closed/uploaded the files and same-BE
uploads may have admitted their segment/index blocks to the local cache.
Dropping `partial` does not call `Rowset::clear_cache`; `ADD_PARTIAL_ROWSET`
has no acknowledgment, and target failures before/after assembly also lack
cleanup for accepted partials. The cloud forwarding receiver has the same gap
because its completed file writers live outside the rowset writer. Repeated
failed loads therefore retain dead cache blocks until eviction. Please retain
cleanup ownership through an acknowledged handoff and clear every abort path;
fence both direct and packed asynchronous cache writes so they cannot
repopulate a cleared key.
##########
be/src/exec/sink/load_stream_stub.cpp:
##########
@@ -123,6 +123,12 @@ int
LoadStreamReplyHandler::on_received_messages(brpc::StreamId id, butil::IOBuf
<< status;
}
}
+ if (response.has_write_context()) {
+ auto writer_id = response.write_context().writer_id();
+ std::lock_guard lock(stub->_write_context_mutex);
+ stub->_write_context_responses[writer_id] = std::move(response);
Review Comment:
[P1] Attach the load memory context before parsing this response. A MOW
write context can contain every historical rowset (with a full schema copied
into each one) plus the aggregated delete bitmap, but `on_received_messages`
runs on a BRPC bthread with no `SCOPED_ATTACH_TASK`; parsing and retaining that
protobuf therefore bypasses the load/query limiter. A high-version tablet,
especially with concurrent or unshared writers, can exhaust BE memory instead
of being limited or cancelled. Carry the load `ResourceContext` into the
stub/handler, attach it at callback entry, and reserve/cap or deduplicate the
snapshot payload before allocation.
--
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]
