feat(storage): only update related read version on flush finish

src/common/src/config.rs

@@ -620,6 +620,7 @@ pub struct StorageConfig {
 
     /// The threshold for the number of immutable memtables to merge to a new imm.
     #[serde(default = "default::storage::imm_merge_threshold")]
+    #[deprecated]
     pub imm_merge_threshold: usize,
 
     /// Whether to enable write conflict detection

src/storage/hummock_test/src/hummock_read_version_tests.rs

@@ -12,6 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+use std::collections::HashMap;
 use std::ops::Bound;
 use std::sync::Arc;
 
@@ -26,6 +27,7 @@ use risingwave_hummock_sdk::key::{key_with_epoch, map_table_key_range};
 use risingwave_hummock_sdk::LocalSstableInfo;
 use risingwave_meta::hummock::test_utils::setup_compute_env;
 use risingwave_pb::hummock::{KeyRange, SstableInfo};
+use risingwave_storage::hummock::event_handler::TEST_LOCAL_INSTANCE_ID;
 use risingwave_storage::hummock::iterator::test_utils::{
     iterator_test_table_key_of, iterator_test_user_key_of,
 };
@@ -48,7 +50,12 @@ async fn test_read_version_basic() {
     let mut epoch = test_epoch(1);
     let table_id = 0;
     let vnodes = Arc::new(Bitmap::ones(VirtualNode::COUNT));
-    let mut read_version = HummockReadVersion::new(TableId::from(table_id), pinned_version, vnodes);
+    let mut read_version = HummockReadVersion::new(
+        TableId::from(table_id),
+        TEST_LOCAL_INSTANCE_ID,
+        pinned_version,
+        vnodes,
+    );
 
     {
         // single imm
@@ -178,7 +185,7 @@ async fn test_read_version_basic() {
             ],
             vec![],
             epoch_id_vec_for_clear,
-            batch_id_vec_for_clear,
+            HashMap::from_iter([(TEST_LOCAL_INSTANCE_ID, batch_id_vec_for_clear)]),
             1,
         ));
 
@@ -267,6 +274,7 @@ async fn test_read_filter_basic() {
     let vnodes = Arc::new(Bitmap::ones(VirtualNode::COUNT));
     let read_version = Arc::new(RwLock::new(HummockReadVersion::new(
         TableId::from(table_id),
+        TEST_LOCAL_INSTANCE_ID,
         pinned_version,
         vnodes.clone(),
     )));