Repository navigation
Conversation
### What problem does this PR solve? Issue Number: N/A Related PR: apache#63258 Problem Summary: Remote broadcast channels share a PBlock but still copy its payload into a separate protobuf request buffer for every destination. Cache the encoded block field before enqueueing and compose each routing header with shared IOBuf references through brpc::SerializedRequest. Reuse the existing wire format, batching, callbacks and HTTP selection, and account the cache once in broadcast queue admission. Preserve required-header-field validation and the 2 GiB protobuf size limit. The preceding standalone prototype reduced packing CPU for 1 MiB x 8 destinations from 3370.81 us to 402.43 us; this is not an end-to-end measurement of this production patch. Receiver work and network bytes are unchanged. ### Release note Reduce repeated protobuf serialization and payload copies for exchange broadcasts to multiple remote destinations. ### Check List (For Author) - Test: Unit Test — 24 ASAN tests passed, including 14 new broadcast transport tests and 10 existing exchange tests. clang-format 16, build hygiene, and clang-tidy checks passed. - Behavior changed: No (SQL results and exchange wire protocol are unchanged) - Does this need documentation: No
Contributor
|
Thank you for your contribution to Apache Doris. Please clearly describe your PR:
|
Contributor
Author
|
/review |
![github-actions[bot]](https://avatars.githubusercontent.com/in/15368?s=60&v=4){kind=small}
Contributor
There was a problem hiding this comment.
Complete static review of PR #68961 at head 70c4f63fe54db120cbc8dc482baa9887d527f2bb against the supplied merge-base diff. No substantiated new findings and no existing P0/P1 inline findings to carry forward. The main review, one independent full review, and two separate wire/lifecycle risk reviews converged in round 1 with NO_NEW_VALUABLE_FINDINGS; all suspicious points were resolved and the final sweep covered all five changed files. No additional user focus was supplied.
The traced flow is producer serialization → immutable cache preparation → queue admission/publication → per-destination header plus shared buffer references → asynchronous RPC → the existing receiver and completion paths.
Critical checkpoint conclusions:
- Goal and evidence: The code encodes each eligible broadcast block once and appends references for each remote destination. Inspected wire-equivalence and backing-pointer assertions directly cover that mechanism. End-to-end speedup is not established by this review.
- Scope and reuse: The five-file change is focused on broadcast protobuf preparation, transport, accounting, and tests. The extracted repeated-block helper preserves the old metadata and field-presence behavior.
- Concurrency and locks: Pipeline producers prepare caches before publication; destination queue/request/sequence state uses
RpcInstance::mutex, and admission counters use the limiter lock. Immutable backing buffers use atomic reference counts. Large cache encoding occurs outside those locks; sends retain the existing destination lock and append references. No new lock-order inversion or blocking wait was identified. - Lifecycle and initialization: A fresh holder owns each source block/cache. RPC requests and controllers retain buffer references after holders die. Fresh per-RPC controller/response/callback state and weak callback ownership preserve reentry and late-completion cleanup without adding a query ownership cycle. The method descriptor uses function-local initialization.
- Configuration: No configuration is added. The existing multi-block query option drives both fragment encoding and request assembly; existing transport, timeout, and overcrowding settings remain in use.
- Compatibility: Singular/repeated field IDs, metadata, compression bytes, routing, packet sequences, and EOS retain their receiver-visible meaning. No storage-format or BE execution-version change is introduced; old receiver parsing remains applicable during rolling upgrades.
- Thrift/protobuf definitions: None are changed. Referenced definitions were inspected to confirm the fragment's nested messages have no required fields; required routing fields are validated in the header.
- Parallel paths: Local-only, regular shuffle, single remote receiver, empty EOS, uncached oversized fragments, and mixed cached/uncached batches retain their existing paths. Any nonempty uncached block excludes the whole batch from the shared-fragment branch.
- Conditions and errors: Cache eligibility, required-header validation, and the aggregate 2 GiB limit have concrete purposes. These replace checks bypassed by brpc's
SerializedRequest. Borrowed fields detach on exit; serialization errors propagate through the existing task/callback failure paths. - Test coverage: The 14 generated unit cases cover wire equivalence, shared backing buffers, concurrent destination encoding/holder release, repeated defaults, batch/EOS boundaries, the actual send branch, single-receiver exclusion, missing required fields, fixed admission accounting, aggregate size rejection, and helper-level asynchronous failure/EOF/expired callbacks. The send-branch test manually drives sends and suppresses production callback chaining. Full production chaining, queue cancellation/EOF dependency signaling, mixed-cache/HTTP-boundary integration, and actual block decompression are not proved by the new tests; static tracing found no introduced defect in those paths.
- Test results: No regression result files change. The unit assertions match their constructed messages. No builds or runtime tests were run here, and the author's ASAN/style/tidy claims were not independently reproduced.
- Observability: Existing execution/serialization/compression/RPC counters and failure logs remain available. New preparation time is included in sink execution time; no new operational state requiring a separate metric was identified.
- Transactions and persistence: No transaction, EditLog, commit/replay, visible-version, or delete-bitmap behavior changes; those invariants are outside this diff.
- Data writes and crash behavior: No persistent data-write path changes. Shared payload ownership, failure propagation, EOF cleanup, and cancellation unblocking were traced without a new atomicity or resource-release defect.
- FE-BE variables: No new transmitted variable or FE path is added.
- Performance and memory: Payload encoding/copying occurs once per eligible holder, with per-destination buffer-reference work. The original payload plus encoded cache is charged once at admission and that fixed debit is released on holder destruction. Outstanding RPC storage follows the existing in-flight lifetime and global
IOBufBlockMemorytracking. The additional queued cache is accounted for; production throughput/latency was not measured. - Other BE checkpoints: The text-only
build-support/check-build-hygiene.shpassed header layering/budgets/PCH, template pairing, and unity-source inclusion checks. The new test is discovered by the existing recursive test glob and includes no source.cpp. No new libc API or dependency/toolchain change was identified; no release ELF ABI audit was performed.
The shared-PBlock mutation in the existing HTTP helper and repeated-block HTTP limitations were checked against the base and dismissed as pre-existing behavior. They do not support a new inline finding in this patch. Static-only review completion does not establish runtime or benchmark results.
Contributor
Author
|
run buildall |
Contributor
Author
|
The earlier standalone prototype reported in this PR measured sender-side protobuf request packing with a 1 MiB payload broadcast to 8 destinations:
This is approximately 8.38× faster, or an 88.06% reduction in packing CPU time. The mechanism is to encode the block field once and append shared buffer references for each destination, avoiding repeated payload encoding/copies. These numbers come from the earlier standalone prototype; they are not a benchmark of the production patch at |
Contributor
TPC-H: Total hot run time: 29282 ms |
Contributor
TPC-DS: Total hot run time: 151861 ms |
Contributor
ClickBench: Total hot run time: 24.73 s |
Contributor
BE Regression && UT Coverage ReportIncrement line coverage Increment coverage report
|
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
What problem does this PR solve?
Related PR: #63258
Remote broadcast channels share the same
PBlock, but brpc normally serializes its largecolumn_valuesinto a separate request buffer for every destination. Broadcasting an S-byte block to N destinations therefore copies roughly N*S bytes while preparing the protobuf requests.Encode the
blockor repeatedblocksfield once inBroadcastPBlockHolder, before publishing it to the destination queues. Each RPC serializes its routing fields and appends references to the cached encoding throughbrpc::SerializedRequest. The receiver continues to parse the existingPTransmitDataParamswire format. Queue order, packet sequences, EOS, and per-RPC callback ownership are preserved.The original PBlock remains available to the existing send paths. Queue admission accounts for both it and the cache once per holder. Single-destination broadcasts and oversized fragments retain the original path; assembled requests retain required-field validation and the protobuf size limit. The HTTP borrowed-data fix remains in #63258.
The earlier standalone prototype measured packing CPU at 1 MiB x 8 destinations dropping from 3370.81 us to 402.43 us. This is motivation for the change, not an end-to-end measurement of this production patch. Network bytes and receiver processing are unchanged.
Release note
Reduce repeated protobuf serialization and payload copies when broadcasting exchange blocks to multiple remote destinations.
Check List (For Author)
broadcast_protobuf_test.cppcovers wire equivalence, shared backing buffers, batching/EOS, the actual send branch, queue accounting, size limits, and asynchronous RPC completion/cancellation.ENABLE_PCH=OFF ENABLE_UNITY_BUILD=ON ./run-be-ut.sh -j 48 --run --filter='*BroadcastProtobufTest*:*ExchangeSinkOperatorTest*:*ExchangeSinkTest*'— 24 tests passed (14 new and 10 existing).build-support/clang-format.sh/build-support/check-format.shwith clang-format 16.0.6;build-support/check-build-hygiene.shandgit diff --checkpassed.build-support/run-clang-tidy.sh: all changed production lines and the entire new test file passed. The local LLVM 20 package needed an explicit resource-directory argument.Check List (For Reviewer who merge this PR)