feat: add skip_wal_replay to OpenRegion instruction (GreptimeTeam#2977)

feat: add skip_wal_replay to OpenRegion instruction

Author: WenyXu

src/common/meta/src/instruction.rs

@@ -98,6 +98,8 @@ pub struct OpenRegion {
     pub region_options: HashMap<String, String>,
     #[serde(default)]
     pub region_wal_options: HashMap<String, String>,
+    #[serde(default)]
+    pub skip_wal_replay: bool,
 }
 
 impl OpenRegion {
@@ -106,12 +108,14 @@ impl OpenRegion {
         path: &str,
         region_options: HashMap<String, String>,
         region_wal_options: HashMap<String, String>,
+        skip_wal_replay: bool,
     ) -> Self {
         Self {
             region_ident,
             region_storage_path: path.to_string(),
             region_options,
             region_wal_options,
+            skip_wal_replay,
         }
     }
 }
@@ -227,12 +231,13 @@ mod tests {
             "test/foo",
             HashMap::new(),
             HashMap::new(),
+            false,
         ));
 
         let serialized = serde_json::to_string(&open_region).unwrap();
 
         assert_eq!(
-            r#"{"OpenRegion":{"region_ident":{"cluster_id":1,"datanode_id":2,"table_id":1024,"region_number":1,"engine":"mito2"},"region_storage_path":"test/foo","region_options":{},"region_wal_options":{}}}"#,
+            r#"{"OpenRegion":{"region_ident":{"cluster_id":1,"datanode_id":2,"table_id":1024,"region_number":1,"engine":"mito2"},"region_storage_path":"test/foo","region_options":{},"region_wal_options":{},"skip_wal_replay":false}}"#,
             serialized
         );
 
@@ -289,6 +294,7 @@ mod tests {
             region_storage_path,
             region_options,
             region_wal_options: HashMap::new(),
+            skip_wal_replay: false,
         };
         assert_eq!(expected, deserialized);
     }

src/datanode/src/heartbeat/handler.rs

@@ -55,7 +55,8 @@ impl RegionHeartbeatResponseHandler {
                 region_storage_path,
                 region_options,
                 region_wal_options,
-            }) => Ok(Box::new(|region_server| {
+                skip_wal_replay,
+            }) => Ok(Box::new(move |region_server| {
                 Box::pin(async move {
                     let region_id = Self::region_ident_to_region_id(&region_ident);
                     // TODO(niebayes): extends region options with region_wal_options.
@@ -64,7 +65,7 @@ impl RegionHeartbeatResponseHandler {
                         engine: region_ident.engine,
                         region_dir: region_dir(&region_storage_path, region_id),
                         options: region_options,
-                        skip_wal_replay: false,
+                        skip_wal_replay,
                     });
                     let result = region_server.handle_request(region_id, request).await;
 
@@ -244,6 +245,7 @@ mod tests {
             path,
             HashMap::new(),
             HashMap::new(),
+            false,
         ))
     }
 

src/meta-srv/src/procedure/region_failover.rs

@@ -622,6 +622,7 @@ mod tests {
                             &path,
                             HashMap::new(),
                             HashMap::new(),
+                            false
                         )))
                         .unwrap(),
                     ))

src/meta-srv/src/procedure/region_failover/activate_region.rs

@@ -91,6 +91,7 @@ impl ActivateRegion {
             &region_storage_path,
             region_options.clone(),
             region_wal_options.clone(),
+            false,
         ));
 
         self.region_storage_path = Some(region_storage_path);
@@ -236,6 +237,7 @@ mod tests {
                     &env.path,
                     HashMap::new(),
                     HashMap::new(),
+                    false
                 )))
                 .unwrap(),
             ))
@@ -307,6 +309,7 @@ mod tests {
                     &env.path,
                     HashMap::new(),
                     HashMap::new(),
+                    false
                 )))
                 .unwrap(),
             ))

src/meta-srv/src/procedure/region_migration/open_candidate_region.rs

@@ -90,6 +90,7 @@ impl OpenCandidateRegion {
             &region_storage_path,
             region_options,
             region_wal_options,
+            true,
         ));
 
         Ok(open_instruction)
@@ -215,6 +216,7 @@ mod tests {
             region_storage_path: "/bar/foo/region/".to_string(),
             region_options: Default::default(),
             region_wal_options: Default::default(),
+            skip_wal_replay: true,
         })
     }