RongtongJin commented on code in PR #5606:
URL: https://github.com/apache/rocketmq/pull/5606#discussion_r1033168438
##########
store/src/main/java/org/apache/rocketmq/store/queue/BatchConsumeQueue.java:
##########
@@ -1005,4 +1005,83 @@ public void cleanSwappedMap(long
forceCleanSwapIntervalMs) {
public MappedFileQueue getMappedFileQueue() {
return mappedFileQueue;
}
+
+ @Override
+ public long estimateMessageCount(long from, long to, MessageFilter filter)
{
+ long physicalOffsetFrom = from * CQ_STORE_UNIT_SIZE;
+ long physicalOffsetTo = to * CQ_STORE_UNIT_SIZE;
+ List<MappedFile> mappedFiles =
mappedFileQueue.range(physicalOffsetFrom, physicalOffsetTo);
+ if (mappedFiles.isEmpty()) {
+ return -1;
+ }
+
+ boolean sample = false;
+ long match = 0;
+ long raw = 0;
+
+ for (MappedFile mappedFile : mappedFiles) {
+ int start = 0;
+ int len = mappedFile.getFileSize();
+ // First file segment
+ if (mappedFile.getFileFromOffset() <= physicalOffsetFrom) {
+ start = (int) (physicalOffsetFrom -
mappedFile.getFileFromOffset());
+ if (mappedFile.getFileFromOffset() + mappedFile.getFileSize()
>= physicalOffsetTo) {
+ // Current mapped file covers search range completely.
+ len = (int) (physicalOffsetTo - physicalOffsetFrom);
+ } else {
+ len = mappedFile.getFileSize() - start;
+ }
+ }
+
+ // Scan partial of the last file segment
+ if (0 == start && mappedFile.getFileFromOffset() +
mappedFile.getFileSize() > physicalOffsetTo) {
+ len = (int) (physicalOffsetTo -
mappedFile.getFileFromOffset());
+ }
+
+ SelectMappedBufferResult slice =
mappedFile.selectMappedBuffer(start, len);
+ if (null != slice) {
+ try {
+ ByteBuffer buffer = slice.getByteBuffer();
+ int current = 0;
+ while (current < len) {
+ // Skip physicalOffset and message length fields.
+ buffer.position(current + 8 + 4);
Review Comment:
It would be better to use constants instead of 4+8, like
BatchConsumeQueue.MSG_TAG_OFFSET_INDEX
##########
store/src/main/java/org/apache/rocketmq/store/queue/BatchConsumeQueue.java:
##########
@@ -1005,4 +1005,83 @@ public void cleanSwappedMap(long
forceCleanSwapIntervalMs) {
public MappedFileQueue getMappedFileQueue() {
return mappedFileQueue;
}
+
+ @Override
+ public long estimateMessageCount(long from, long to, MessageFilter filter)
{
+ long physicalOffsetFrom = from * CQ_STORE_UNIT_SIZE;
+ long physicalOffsetTo = to * CQ_STORE_UNIT_SIZE;
+ List<MappedFile> mappedFiles =
mappedFileQueue.range(physicalOffsetFrom, physicalOffsetTo);
+ if (mappedFiles.isEmpty()) {
+ return -1;
+ }
+
+ boolean sample = false;
+ long match = 0;
+ long raw = 0;
+
+ for (MappedFile mappedFile : mappedFiles) {
+ int start = 0;
+ int len = mappedFile.getFileSize();
+ // First file segment
+ if (mappedFile.getFileFromOffset() <= physicalOffsetFrom) {
+ start = (int) (physicalOffsetFrom -
mappedFile.getFileFromOffset());
+ if (mappedFile.getFileFromOffset() + mappedFile.getFileSize()
>= physicalOffsetTo) {
+ // Current mapped file covers search range completely.
+ len = (int) (physicalOffsetTo - physicalOffsetFrom);
+ } else {
+ len = mappedFile.getFileSize() - start;
+ }
+ }
+
+ // Scan partial of the last file segment
+ if (0 == start && mappedFile.getFileFromOffset() +
mappedFile.getFileSize() > physicalOffsetTo) {
+ len = (int) (physicalOffsetTo -
mappedFile.getFileFromOffset());
+ }
+
+ SelectMappedBufferResult slice =
mappedFile.selectMappedBuffer(start, len);
+ if (null != slice) {
+ try {
+ ByteBuffer buffer = slice.getByteBuffer();
+ int current = 0;
+ while (current < len) {
+ // Skip physicalOffset and message length fields.
+ buffer.position(current + 8 + 4);
+ long tagCode = buffer.getLong();
Review Comment:
Since there may be multiple messages in one BCQ index, IMO, It would be
better to consider using the batch size field to estimate a more accurate value.
--
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]
