[TASK-140] Priority Queue for downloading remote segments #187
+1,629
−145
Conversation
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
fresh-borzoni
force-pushed
the
prioriry-prefetch-remote-log
branch
2 times, most recently
from
f183f8a to
39733e9
Compare
Contributor
Author
There was a problem hiding this comment.
Pull request overview
This PR implements remote log segment downloading with priority-based scheduling for the Rust client, matching the Java client's behavior. The implementation addresses issue #140 by adding a priority queue for downloading remote segments with concurrency controls.
Changes:
- Added RemoteLogDownloader coordinator with event-driven architecture using priority queue (BinaryHeap) for download scheduling
- Implemented two-layer concurrency control: download threads limit (default: 3) and prefetch limit (default: 4)
- Added file-backed streaming for remote log segments to avoid loading entire files into memory, using RAII guards for cleanup
- Extended protobuf schema to include max_timestamp field for priority ordering
- Added exponential backoff retry logic for failed downloads (100ms to 5s with jitter)
Reviewed changes
Copilot reviewed 9 out of 9 changed files in this pull request and generated 4 comments.
Show a summary per file
| File | Description |
|---|---|
| crates/fluss/src/client/table/remote_log.rs | Core implementation of RemoteLogDownloader with priority queue, coordinator loop, and retry logic |
| crates/fluss/src/record/arrow.rs | Added LogRecordsSource trait with MemorySource and FileSource implementations for file-backed streaming |
| crates/fluss/src/client/table/log_fetch_buffer.rs | Added RemoteCompletedFetch and RemotePendingFetch implementations |
| crates/fluss/src/proto/fluss_api.proto | Added optional max_timestamp field to PbRemoteLogSegment |
| crates/fluss/src/config.rs | Added scanner_remote_log_prefetch_num and scanner_remote_log_download_threads config options |
| crates/fluss/src/client/table/scanner.rs | Updated LogFetcher initialization to pass config parameters |
| crates/fluss/src/client/connection.rs | Added config() getter method |
| crates/fluss/src/client/table/mod.rs | Exported new constants |
| crates/fluss/src/util/mod.rs | Removed delete_file() function (replaced by RAII) |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
fresh-borzoni
force-pushed
the
prioriry-prefetch-remote-log
branch
from
39733e9 to
bec7bd2
Compare
Contributor
Author
|
rebased and fixed failing test |
zhaohaidao
reviewed
Jan 21, 2026
| Cancelled, | ||
| } | ||
|
|
||
| /// Production implementation of RemoteLogFetcher that downloads from actual storage |
Contributor
There was a problem hiding this comment.
What exactly does "Production" mean here?
Contributor
Author
There was a problem hiding this comment.
it means that it has counterpart struct FakeFetcher used in testing
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.
Linked issue: close #140
This PR implements remote log segment downloading for the Rust client with priority-based scheduling and concurrency limiting, matching the Java client's behavior.
Change log
RemoteLogDownloader: Background coordinator that manages concurrent downloads of remote log segments from S3/filesystem storagePriorityBlockingQueueorderingscanner_remote_log_prefetch_numandscanner_remote_log_download_threadsconfig options matching Java client defaultsFetchLogRequestto includeremote_log_fetch_infofor remote segment metadataAPI and Format
Configuration API additions:
Config::scanner_remote_log_prefetch_num(default: 4)Config::scanner_remote_log_download_threads(default: 3)No breaking changes to existing public APIs.
Note:
The Rust implementation uses an event-driven actor model with a single coordinator task that owns all download state, eliminating the need for locks.
Unlike Java's thread pool approach with blocking threads and synchronization, Rust uses
async tasks and a
select!loop that reacts to events (NewRequest, DownloadFinished, RecycleNotification). Additionally, Rust implements automatic retry with exponential backoff (100ms to 5s with jitter) for failed downloads, it's different from Java, but we need it to not thrash the PQ as it would be stuck on the same prioritised failed task and try to make progress. Worth discussing though!