[improvement](compaction) Support preferred peer reads - #67375
Conversation
|
Thank you for your contribution to Apache Doris. Please clearly describe your PR:
|
There was a problem hiding this comment.
Warning
Copilot couldn't run its full agentic review because it didn't start before the timeout. Make sure your repository has a runner available, or add a copilot-code-review.yml file specifying one with the runs-on attribute. See the docs for more details.
Pull request overview
Adds support for “preferred peer” reads in compaction/rowset reading by propagating a per-request peer address through RowsetReader into IOContext, allowing a worker to attempt a specific coordinator peer before falling back to remote storage.
Changes:
- Extend
RowsetReaderAPI to set a preferred file-cache peer (host/port) and propagate viaIOContext. - Update
CachedRemoteFileReaderto attempt the preferred peer even when global peer reads are disabled, with fallback to remote on failure. - Add/adjust unit tests to verify propagation and preferred-peer behavior.
Reviewed changes
Copilot reviewed 8 out of 8 changed files in this pull request and generated 6 comments.
Show a summary per file
| File | Description |
|---|---|
| be/src/storage/rowset/rowset_reader.h | Adds new RowsetReader API for preferred peer. |
| be/src/storage/rowset/beta_rowset_reader.h | Implements preferred peer propagation into read options’ IOContext. |
| be/src/io/io_common.h | Adds preferred peer fields to IOContext. |
| be/src/io/cache/cached_remote_file_reader.cpp | Implements preferred-peer read attempt and remote fallback. |
| be/test/storage/rowset/beta_rowset_test.cpp | Adds test asserting preferred peer propagates into segment IO context. |
| be/test/io/cache/cached_remote_file_reader_peer_test.cpp | Updates/adds tests for preferred peer behavior and fallback. |
| be/test/storage/iterator/block_reader_binlog_vcollect_merge_test.cpp | Updates fake reader to satisfy new RowsetReader API. |
| be/test/storage/index/inverted/similarity/collection_statistics_test.cpp | Updates mock reader to satisfy new RowsetReader API. |
💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.
|
/review |
![github-actions[bot]](https://avatars.githubusercontent.com/in/15368?s=60&v=4){kind=small}
There was a problem hiding this comment.
Review summary
Status: capped/incomplete. The focused per-read preferred-peer branch and direct remote fallback work, but six distinct issues still prevent the preferred endpoint from being a complete, bounded, and observable compaction contract. Two accepted issues were found in the third and final permitted round, so the substantiated set below is submitted, but further convergence was not allowed.
Critical checkpoint conclusions
- Goal and proof: Partially achieved. The tests prove preferred-peer success, single-read fallback, and propagation into the ordinary lazy segment context. They do not prove end-to-end compaction routing: footer pre-reads, standalone index compaction, grouped range readers, and variant metadata phases bypass or lose the endpoint.
- Scope and clarity: The diff is small and the local routing branch is clear, but storing task routing only in one
BetaRowsetReader's private read options is too narrow for the existing compaction topology. - Concurrency and thread safety: No new shared mutable endpoint, lock ordering, or background ownership was introduced. The setter is used as setup state and no new race/deadlock was found.
- Lifecycle: The main defects are lifecycle boundaries: IO before reader construction, IO after reconstructing independent index readers, replacement of configured readers per grouped range, and variant scans that unwrap or precede readers. The clone-only concern was dismissed because no current production clone-after-set path exists.
- Configuration: No new configuration was added. Existing cloud/tablet/warmup/bypass and remote-only-on-miss guards behave as intended, and the per-request endpoint correctly overrides the global peer switch. The already-raised invalid-port thread remains the duplicate fence for static endpoint validation.
- Compatibility: No storage-format, persistence, wire-protocol, or FE/BE compatibility change was introduced. All in-tree
RowsetReaderimplementations/test doubles implement the new pure virtual method, and header-hygiene checks pass. - Parallel paths: Not complete. Full vertical pre-reads, V1/V2/SNII index compaction, grouped cloud compaction, and flat/nested variant metadata paths each need explicit task-level propagation.
- Conditions and error handling: Per-read success and S3 fallback cleanup are correct. Timeout-class failures remain unbounded across later uncached reads; NOT_FOUND/overload semantics should remain distinguished.
- Test coverage: Current unit tests cover the narrow happy path and one fallback only. Missing coverage includes cold full vertical compaction, row-count fallback, cold index compaction, grouped segment ranges, flat/nested variant metadata, repeated runtime outage, and compaction statistics.
- Test results:
build-support/check-build-hygiene.shandgit diff --checkpass; live CI reports formatter/checkstyle and repository checks passing. Builds and BE/unit/regression tests were not run in this review-only job, and the build/test CI jobs are skipped. - Observability: Generic file-cache peer counters/hosts are correct on success, but peer bytes/time disappear from merger, task-tracker, tablet-time, and compaction-metric aggregation. Failed-attempt accounting is already covered by the existing inline thread.
- Transactions, persistence, and data writes: No transaction, EditLog, visibility, delete-bitmap, output atomicity, or persistent metadata behavior is changed by this PR.
- FE/BE variables: No new FE-to-BE or serialized variable is introduced.
- Performance: Runtime-unreachable preferred peers can add the 5-second RPC timeout to every later uncached operation, while the propagation gaps send cold compaction reads back to object storage and defeat the intended optimization.
- Other checks: No new memory-safety, nullable, data-version, or security-sensitive behavior was identified. No additional user-provided review focus was supplied.
All six live pre-existing inline threads were treated as hard duplicate fences; the comments in this review are distinct.
| std::vector<RowwiseIteratorUPtr>* out_iters, | ||
| bool use_cache = false) override; | ||
| void reset_read_options() override; | ||
| void set_preferred_file_cache_peer(const std::string& host, int32_t port) override { |
There was a problem hiding this comment.
[P1] Route compaction's pre-reader footer reads through this endpoint
With the default compaction_batch_size == -1, Merger::vertical_merge_rowsets() calls BetaRowset::load_segments() for first-time footer sampling before any get_segment_iterators() call. That overload opens segments with a null IOContext, so on the cold distributed worker this feature targets, with global peer reads disabled, those footer reads still go to object storage even after this setter was called. The row-ID-conversion fallback similarly calls get_segment_num_rows(..., _stats) without _read_options.io_ctx. Please plumb the preference into these pre-reads and cover cold-cache vertical compaction; the new unit test exercises only the later lazy-iterator path.
There was a problem hiding this comment.
later pr will set peer info
| bool use_cache = false) override; | ||
| void reset_read_options() override; | ||
| void set_preferred_file_cache_peer(const std::string& host, int32_t port) override { | ||
| _read_options.io_ctx.preferred_peer_host = host; |
There was a problem hiding this comment.
[P1] Propagate the preferred peer into index compaction
do_inverted_index_compaction() reconstructs IndexFileReaders directly from the source rowsets, then calls init(), open(), and open_snii_index() without an IOContext. With global peer reads disabled, cold V1/V2/SNII source-index reads therefore still use object storage even after the rowset reader was configured. This phase runs after the row merge and does not reuse input_rs_readers, so fixing the footer pre-reads does not cover it. Please carry a compaction-level preferred-peer context into every source index-container read and add a cold-cache index-compaction test.
There was a problem hiding this comment.
later pr will set peer info
| void reset_read_options() override; | ||
| void set_preferred_file_cache_peer(const std::string& host, int32_t port) override { | ||
| _read_options.io_ctx.preferred_peer_host = host; | ||
| _read_options.io_ctx.preferred_peer_port = port; |
There was a problem hiding this comment.
[P1] Preserve the preference when grouped compaction rebuilds readers
The production grouped path in CloudCumulativeCompaction::do_merge_input_rowsets() ignores the configured input_rs_readers vector and calls input_rowset->create_reader() for every segment range. Each fresh reader has an empty preferred endpoint, so with global peer reads disabled the range's normal segment footer/index/data IO goes to object storage. This is not the clone or pre-reader-footer case: segment_range skips that sampler and the configured reader is replaced outright. Please keep this routing state at compaction-task scope and apply it to every range reader, with a cold-cache grouped-compaction test.
There was a problem hiding this comment.
later pr will set peer info
| @@ -58,6 +59,7 @@ class RowsetReader { | |||
| std::vector<RowwiseIteratorUPtr>* out_iters, | |||
| bool use_cache = false) = 0; | |||
| virtual void reset_read_options() = 0; | |||
There was a problem hiding this comment.
[P1] Route variant-compaction metadata reads through the task peer
With enable_vertical_compact_variant_subcolumns enabled by default, both compaction build_basic_info() paths call get_extended_compaction_schema() before the input readers are created. Its variant aggregation opens segments, column metadata, and external metadata with null/default contexts. Nested-group streaming-plan construction unwraps configured readers back to rowsets and repeats those context-free reads, and validation can do so again. On a cold worker these source reads bypass the preferred coordinator. Please keep the endpoint in task-level IO state and pass it through variant schema discovery, streaming planning, and validation, with cold-cache flat/nested variant tests.
There was a problem hiding this comment.
later pr will set peer info
| return _execute_s3_fallback(empty_start, span_size, buffer, peer_result, stats, io_ctx); | ||
| } | ||
|
|
||
| if (!io_ctx->preferred_peer_host.empty()) { |
There was a problem hiding this comment.
[P1] Bound retries after the preferred peer becomes unreachable
This branch retains the endpoint after every RPC error, so each later uncached read can wait the 5-second BRPC timeout before falling back to S3. The ordinary candidate path evicts an endpoint after peer_rpc_failure_eviction_threshold consecutive failures (default 3), but the preferred branch bypasses that protection. A coordinator outage can therefore add roughly five seconds per cold operation throughout a compaction. Please add a task-scoped circuit breaker/cooldown for timeout-class RPC failures while preserving cache-miss/overload semantics, and add a multi-read outage test. This is distinct from the existing zero-port thread because the endpoint is initially valid and fails at runtime.
| // option and the global async file-cache write switch. | ||
| std::optional<CacheWriteMode> cache_write_mode_override = std::nullopt; | ||
| // Optional per-request peer; try only this peer before falling back to remote storage. | ||
| std::string preferred_peer_host {}; |
There was a problem hiding this comment.
[P2] Preserve preferred-peer IO in compaction statistics
A successful preferred read is recorded in FileCacheStatistics::bytes_read_from_peer and peer_io_timer, but both merger paths export only local/remote bytes and time, and Compaction updates only the local/remote compaction metrics. With global peer reading disabled, this feature therefore moves input bytes out of the remote category without adding them anywhere in the compaction task/tracker statistics, so reported compaction IO drops despite the same data being read. Please propagate peer bytes/time through horizontal, vertical, and grouped compaction (or deliberately classify them into an existing category) and add a statistics assertion.
|
run buildall |
TPC-H: Total hot run time: 16811 ms |
TPC-DS: Total hot run time: 82433 ms |
ClickBench: Total hot run time: 14.68 s |
### What problem does this PR solve? Issue Number: None Related PR: apache#67466 Problem Summary: Distributed compaction readers need to read cached data from the coordinator even when general peer reads are disabled. Carry an optional preferred peer through RowsetReader and IOContext, try only that peer first, and fall back to remote storage on failure. Because worker compaction reads are one-shot, skip worker file-cache writeback and ineffective prefetch when a preferred peer is configured. Record successful preferred peers in PeerCacheNodes and cover routing, fallback, propagation, and no-write behavior. ### Release note Support per-request preferred peer reads for backend rowset readers. Distributed compaction workers do not cache input blocks fetched through the preferred-peer path. ### Check List (For Author) - Test: Unit Test not completed - ./run-be-ut.sh --run --filter=CachedRemoteFileReaderPeerTest.read_at_uses_preferred_peer_when_global_peer_read_disabled -j4 was attempted previously; configuration failed because thirdparty/installed/arrow-24.0.0 is incomplete - Per request, the merged changes were not compiled or run - Behavior changed: Yes (a configured preferred peer is tried before remote storage, reported in PeerCacheNodes, and does not write worker file cache) - Does this need documentation: No
Later, we will support distributed parallel compaction. compaction workers may not have the input rowsets in their local file cache.
This PR allows workers to read cached input data from the coordinator when available. It propagates an optional preferred peer through RowsetReader and IOContext. When configured, the worker tries only that peer and falls back directly to remote storage if the peer read fails.
related pr:
#67360