feat: various soft-drop improvements

- Bump `greptime-proto` dependency revision
  (`Cargo.toml`, `Cargo.lock`)
- Pass `region_wal_options` directly without pre-serialization in undrop flow
  (`src/common/meta/src/ddl/undrop_table.rs`, `src/common/meta/src/key.rs`)
- Remove unused `RegionNumber` import
  (`src/common/meta/src/ddl/utils.rs`)
- Add `reset_failure_detectors` to test mock
  (`src/common/meta/src/ddl/tests/drop_table.rs`)
- Add JSON roundtrip tests for `UndropTableTask` and `PurgeDroppedTableTask`
  (`src/common/meta/src/rpc/ddl.rs`)

Signed-off-by: Lei, HUANG <ratuthomm@gmail.com>

Cargo.lock

@@ -5938,7 +5938,7 @@ dependencies = [
 [[package]]
 name = "greptime-proto"
 version = "0.1.0"
-source = "git+https://github.com/GreptimeTeam/greptime-proto.git?rev=b6b665d8b6fce4f1f0da49c8c79fb150636df4ff#b6b665d8b6fce4f1f0da49c8c79fb150636df4ff"
+source = "git+https://github.com/GreptimeTeam/greptime-proto.git?rev=d21d344eb357fd250e0d71b354c8d1b38a077b14#d21d344eb357fd250e0d71b354c8d1b38a077b14"
 dependencies = [
  "prost 0.14.1",
  "prost-types 0.14.1",

Cargo.toml

@@ -158,7 +158,7 @@ fs2 = "0.4"
 fst = "0.4.7"
 futures = "0.3"
 futures-util = "0.3"
-greptime-proto = { git = "https://github.com/GreptimeTeam/greptime-proto.git", rev = "b6b665d8b6fce4f1f0da49c8c79fb150636df4ff" }
+greptime-proto = { git = "https://github.com/GreptimeTeam/greptime-proto.git", rev = "d21d344eb357fd250e0d71b354c8d1b38a077b14" }
 hex = "0.4"
 http = "1"
 humantime = "2.1"

src/common/meta/src/ddl/tests/drop_table.rs

@@ -1041,6 +1041,10 @@ impl RegionFailureDetectorController for RecordingRegionFailureDetectorControlle
         self.registered.lock().await.extend(detecting_regions);
     }
 
+    async fn reset_failure_detectors(&self, detecting_regions: Vec<DetectingRegion>) {
+        self.registered.lock().await.extend(detecting_regions);
+    }
+
     async fn deregister_failure_detectors(&self, detecting_regions: Vec<DetectingRegion>) {
         self.deregistered.lock().await.extend(detecting_regions);
     }

src/common/meta/src/ddl/undrop_table.rs

@@ -213,14 +213,6 @@ pub(crate) async fn open_regions(
     let template = build_template_from_raw_table_info(table_info)?;
     let builder = CreateRequestBuilder::new(template, None);
     let storage_path = region_storage_path(&table_name.catalog_name, &table_name.schema_name);
-    let wal_options = region_wal_options
-        .iter()
-        .map(|(region_number, wal_options)| {
-            serde_json::to_string(wal_options)
-                .map(|wal_options| (*region_number, wal_options))
-                .context(error::SerdeJsonSnafu)
-        })
-        .collect::<Result<HashMap<_, _>>>()?;
     let mut seen_peer_ids = HashSet::new();
     let peers = find_leaders(region_routes)
         .into_iter()
@@ -237,9 +229,9 @@ pub(crate) async fn open_regions(
             let create_request = builder.build_one(
                 region_id,
                 storage_path.clone(),
-                &wal_options,
+                region_wal_options,
                 &HashMap::new(),
-            );
+            )?;
             let request = RegionRequest {
                 header: Some(RegionRequestHeader {
                     tracing_context: TracingContext::from_current_span().to_w3c(),

src/common/meta/src/ddl/utils.rs

@@ -37,7 +37,7 @@ use snafu::{OptionExt, ResultExt, ensure};
 use store_api::metadata::ColumnMetadata;
 use store_api::metric_engine_consts::{LOGICAL_TABLE_METADATA_KEY, MANIFEST_INFO_EXTENSION_KEY};
 use store_api::region_engine::RegionManifestInfo;
-use store_api::storage::{RegionId, RegionNumber};
+use store_api::storage::RegionId;
 use table::metadata::{TableId, TableInfo};
 use table::table_reference::TableReference;
 

src/common/meta/src/key.rs

@@ -1812,15 +1812,11 @@ mod tests {
         let table_metadata_manager = TableMetadataManager::new(mem_kv.clone());
         let task = test_create_table_task(table_name, table_id);
         let table_info = task.table_info.clone();
-        let serialized_options = region_wal_options
-            .iter()
-            .map(|(k, v)| (*k, serde_json::to_string(v).unwrap()))
-            .collect::<HashMap<_, _>>();
         table_metadata_manager
             .create_table_metadata(
                 table_info.clone(),
                 TableRouteValue::physical(region_routes),
-                serialized_options,
+                region_wal_options.clone(),
             )
             .await
             .unwrap();

src/common/meta/src/rpc/ddl.rs

@@ -1948,6 +1948,36 @@ mod tests {
         assert!(matches!(de, DdlTask::PurgeDroppedTable(task) if task == expected));
     }
 
+    #[test]
+    fn test_undrop_table_task_json_roundtrip() {
+        let task = UndropTableTask {
+            catalog: "greptime".to_string(),
+            schema: "public".to_string(),
+            table: "foo".to_string(),
+            table_id: 1024,
+        };
+
+        let output = serde_json::to_vec(&task).unwrap();
+
+        let de = serde_json::from_slice(&output).unwrap();
+        assert_eq!(task, de);
+    }
+
+    #[test]
+    fn test_purge_dropped_table_task_json_roundtrip() {
+        let task = PurgeDroppedTableTask {
+            catalog: "greptime".to_string(),
+            schema: "public".to_string(),
+            table: "foo".to_string(),
+            table_id: Some(1024),
+        };
+
+        let output = serde_json::to_vec(&task).unwrap();
+
+        let de = serde_json::from_slice(&output).unwrap();
+        assert_eq!(task, de);
+    }
+
     #[test]
     fn test_sort_columns() {
         // construct RawSchema