From c91da1777c8724fcfb2261551b75ce9dea5d9e53 Mon Sep 17 00:00:00 2001
From: Patrick Huang <hzxa21@hotmail.com>
Date: Tue, 5 Mar 2024 14:54:46 +0000
Subject: [PATCH] address comments

---
 .../src/hummock/store/hummock_storage.rs      | 31 ++++++++++++++++++-
 1 file changed, 30 insertions(+), 1 deletion(-)

diff --git a/src/storage/src/hummock/store/hummock_storage.rs b/src/storage/src/hummock/store/hummock_storage.rs
index 76b7b56cd04dd..88c123ca5d5bc 100644
--- a/src/storage/src/hummock/store/hummock_storage.rs
+++ b/src/storage/src/hummock/store/hummock_storage.rs
@@ -313,6 +313,7 @@ impl HummockStorage {
             get_committed_read_version_tuple((**pinned_version).clone(), table_id, key_range, epoch)
         } else {
             let vnode = vnode(&key_range);
+            let mut matched_replicated_read_version_cnt = 0;
             let read_version_vec = {
                 let read_guard = self.read_version_mapping.read();
                 read_guard
@@ -321,7 +322,17 @@ impl HummockStorage {
                         v.values()
                             .filter(|v| {
                                 let read_version = v.read();
-                                !read_version.is_replicated() && read_version.contains(vnode)
+                                if read_version.contains(vnode) {
+                                    if read_version.is_replicated() {
+                                        matched_replicated_read_version_cnt += 1;
+                                        false
+                                    } else {
+                                        // Only non-replicated read version with matched vnode is considered
+                                        true
+                                    }
+                                } else {
+                                    false
+                                }
                             })
                             .cloned()
                             .collect_vec()
@@ -332,6 +343,24 @@ impl HummockStorage {
             // When the system has just started and no state has been created, the memory state
             // may be empty
             if read_version_vec.is_empty() {
+                if matched_replicated_read_version_cnt > 0 {
+                    tracing::warn!(
+                        "Read(table_id={} vnode={} epoch={}) is not allowed on replicated read version ({} found). Fall back to committed version (epoch={})",
+                        table_id,
+                        vnode.to_index(),
+                        epoch,
+                        matched_replicated_read_version_cnt,
+                        pinned_version.max_committed_epoch()
+                    );
+                } else {
+                    tracing::debug!(
+                        "No read version found for read(table_id={} vnode={} epoch={}). Fall back to committed version (epoch={})",
+                        table_id,
+                        vnode.to_index(),
+                        epoch,
+                        pinned_version.max_committed_epoch()
+                    );
+                }
                 get_committed_read_version_tuple(
                     (**pinned_version).clone(),
                     table_id,