Repository navigation
[ISSUE #4562]✨Enhance IndexService: improve file loading, add index type support, and optimize file deletion logic - #4563
Conversation
…ype support, and optimize file deletion logic
|
🔊@mxsm 🚀Thanks for your contribution🎉！ 💡CodeRabbit(AI) will review your code first🔥！ Note 🚨The code review suggestions from CodeRabbit are to be used as a reference only, and the PR submitter can decide whether to make changes based on their own judgment. Ultimately, the project management personnel will conduct the final code review💥. |
rocketmq-rust-bot
added
auto merge
ready to review
waiting-review
WalkthroughTwo files modified: constants added to MessageConst for index typing, and IndexService enhanced with type-aware query support, improved file lifecycle management (loading, deletion, destruction), and helper methods for type-prefixed key construction. Tests added for new functionality. Changes
Estimated code review effort🎯 4 (Complex) | ⏱️ ~50 minutes
Poem
Pre-merge checks and finishing touches❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✨ Finishing touches
🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
There was a problem hiding this comment.
Pull request overview
This PR enhances the IndexService implementation with improved file loading logic, support for index types (particularly tag indexing), and optimized file deletion mechanisms. The changes refactor several methods to use more idiomatic Rust patterns and improve concurrency through better lock management.
Key Changes
- Added index type constants (INDEX_KEY_TYPE, INDEX_UNIQUE_TYPE, INDEX_TAG_TYPE) to support different index formats
- Introduced tag-based indexing capability with
query_offset_with_typemethod and corresponding build_index logic - Optimized delete_expired_file logic with two-phase locking strategy to reduce write lock contention
Reviewed changes
Copilot reviewed 2 out of 2 changed files in this pull request and generated 2 comments.
| File | Description |
|---|---|
| rocketmq-common/src/common/message.rs | Adds three index type constants (K, U, T) to support different index key formats |
| rocketmq-store/src/index/index_service.rs | Major refactoring including: improved load method with better error handling, optimized file deletion with two-phase locking, new query_offset_with_type method for tag queries, tag indexing support in build_index, cleaner put_key/retry logic, better lock management in get_and_create_last_index_file, and comprehensive test coverage |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
| /// * `index_type` - Optional index type (e.g., MessageConst::INDEX_TAG_TYPE) If None or empty, | ||
| /// uses default format: topic#key If INDEX_TAG_TYPE, uses format: topic#T#key |
There was a problem hiding this comment.
The documentation formatting is unclear. The sentence runs together without proper punctuation. Consider reformatting to:
index_type- Optional index type (e.g., MessageConst::INDEX_TAG_TYPE). If None or empty, uses default format: topic#key. If INDEX_TAG_TYPE, uses format: topic#T#key
| /// * `index_type` - Optional index type (e.g., MessageConst::INDEX_TAG_TYPE) If None or empty, | |
| /// uses default format: topic#key If INDEX_TAG_TYPE, uses format: topic#T#key | |
| /// * `index_type` - Optional index type (e.g., MessageConst::INDEX_TAG_TYPE). | |
| /// - If None or empty, uses default format: `topic#key`. | |
| /// - If INDEX_TAG_TYPE, uses format: `topic#T#key`. |
| error!( | ||
| "putKey error commitlog {} tags {}", | ||
| dispatch_request.commit_log_offset, tags | ||
| ); |
There was a problem hiding this comment.
Inconsistent error handling: when tag indexing fails, the function continues execution without returning early, unlike the error handling for uniq_key (line 315) and regular keys (line 334). Consider adding a return statement after the error log if tag indexing failures should also stop the indexing process, or add a comment explaining why tag indexing failures are non-fatal.
| ); | |
| ); | |
| return; |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 0
🧹 Nitpick comments (3)
rocketmq-store/src/index/index_service.rs (3)
186-190: Consider usingHashSetfor O(n) lookup instead of O(n*m).The current implementation uses
Vec::contains()insideretain(), resulting in O(n*m) complexity. For large numbers of expired files, aHashSetwould provide O(n+m) complexity.+use std::collections::HashSet; + - let mut destroyed_files = Vec::new(); + let mut destroyed_files = HashSet::new(); for file in files.iter() { let file_name = file.get_file_name(); if file.destroy(3000) { - destroyed_files.push(file_name.clone()); + destroyed_files.insert(file_name.clone()); info!("Delete expired index file success: {}", file_name); } else { error!("Delete expired index file failed: {}", file_name); break; } }
236-272: Consider using.iter().rev().enumerate()for more idiomatic iteration.The current loop pattern
for i in (1..=len).rev()withindex_file_list[i - 1]works but is unconventional. A more idiomatic approach would use the iterator directly.- for i in (1..=index_file_list.len()).rev() { - let f = &index_file_list[i - 1]; - let last_file = i == index_file_list.len(); + let len = index_file_list.len(); + for (rev_idx, f) in index_file_list.iter().rev().enumerate() { + let last_file = rev_idx == 0;
566-568: The 10ms sleep may cause test flakiness in CI environments.The
spawn_blockingtask for flushing may not complete within 10ms on slower or loaded systems. Consider using a synchronization mechanism or accepting that this is a smoke test for the logic flow.If the intent is just to verify no panics occur, the sleep can be removed. If you need to verify the async task completes, consider using a longer timeout or a proper synchronization mechanism:
- // Give async tasks time to complete - tokio::time::sleep(tokio::time::Duration::from_millis(10)).await; + // Give async tasks time to complete - use longer timeout for CI stability + tokio::time::sleep(tokio::time::Duration::from_millis(100)).await;
📜 Review details
Configuration used: Path: .coderabbit.yaml
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (2)
rocketmq-common/src/common/message.rs(1 hunks)rocketmq-store/src/index/index_service.rs(9 hunks)
⏰ Context from checks skipped due to timeout of 90000ms. You can increase the timeout in your CodeRabbit configuration to a maximum of 15 minutes (900000ms). (6)
- GitHub Check: Build & Test (ubuntu-latest)
- GitHub Check: Build & Test (macos-latest)
- GitHub Check: Build & Test (windows-latest)
- GitHub Check: Code Coverage
- GitHub Check: Agent
- GitHub Check: auto-approve
🔇 Additional comments (9)
rocketmq-common/src/common/message.rs (1)
595-599: LGTM!The new index type constants are well-defined and follow the existing naming conventions in
MessageConst. The single-character type codes ("K", "U", "T") are consistent with typical RocketMQ conventions.rocketmq-store/src/index/index_service.rs (8)
129-135: LGTM!Clean, idiomatic implementation that correctly calculates total size based on uniform file sizes.
193-200: LGTM!Clean implementation that safely destroys all index files under a write lock and clears the list.
339-364: LGTM!Tag indexing correctly extracts tags from
properties_mapand uses the type-prefixed key format. Error handling is consistent with the existing pattern.
372-391: LGTM!The refactored
put_keymethod cleanly handles the retry logic with a simple loop that exits on success or when index file creation fails.
414-468: LGTM!The two-phase read-then-write locking approach effectively minimizes lock contention. Using
spawn_blockingfor the flush operation is appropriate since it involves I/O.
470-489: LGTM!Clean implementation using let-else for early return. The conditional checkpoint update based on
is_write_full()is correct behavior.
492-500: LGTM!Simple and well-defined helper functions for consistent key formatting. The inline annotation is appropriate for these small functions.
115-123: The review comment is based on a false premise. Theload()method returnsvoid(unit type), not aResultorbool. There is no return value to "ignore." The suggested diff would not compile because you cannot useif !on a void method.The actual error handling in the code uses
.unwrap()on line 73 ofindex_header.rs, which will panic ifget_bytes()fails. If error handling is a concern, it should address the aggressive.unwrap()rather than a non-existent return value.Likely an incorrect or invalid review comment.
Codecov Report❌ Patch coverage is
Additional details and impacted files@@ Coverage Diff @@
## main #4563 +/- ##
==========================================
+ Coverage 30.79% 31.01% +0.22%
==========================================
Files 673 673
Lines 97983 98095 +112
==========================================
+ Hits 30172 30424 +252
+ Misses 67811 67671 -140 ☔ View full report in Codecov by Sentry. 🚀 New features to boost your workflow:
|
rocketmq-rust-bot
left a comment
rocketmq-rust-bot
left a comment
There was a problem hiding this comment.
LGTM - All CI checks passed ✅
rocketmq-rust-bot
added
approved
Which Issue(s) This PR Fixes(Closes)
Fixes #4562
Brief Description
How Did You Test This Change?
Summary by CodeRabbit
New Features
Improvements
✏️ Tip: You can customize this high-level summary in your review settings.