refactor(meta/watcher): just use Change as event type

Author: drmingdrmer

src/meta/api/src/schema_api_impl.rs

@@ -987,7 +987,6 @@ impl<KV: KVApi> SchemaApi for KV {
                     if_then: vec![
                         // Changing a table in a db has to update the seq of db_meta,
                         // to block the batch-delete-tables when deleting a db.
-                        // TODO: test this when old metasrv is replaced with kv-txn based SchemaApi.
                         txn_op_put(&DatabaseId { db_id }, serialize_struct(&db_meta)?), /* (db_id) -> db_meta */
                         txn_op_put(&dbid_tbname, serialize_u64(table_id)?), /* (tenant, db_id, tb_name) -> tb_id */
                         txn_op_put(&tbid, serialize_struct(&req.table_meta)?), /* (tenant, db_id, tb_id) -> tb_meta */
@@ -1149,7 +1148,6 @@ impl<KV: KVApi> SchemaApi for KV {
                     if_then: vec![
                         // Changing a table in a db has to update the seq of db_meta,
                         // to block the batch-delete-tables when deleting a db.
-                        // TODO: test this when old metasrv is replaced with kv-txn based SchemaApi.
                         txn_op_put(&DatabaseId { db_id }, serialize_struct(&db_meta)?), /* (db_id) -> db_meta */
                         txn_op_del(&dbid_tbname), // (db_id, tb_name) -> tb_id
                         txn_op_put(&tbid, serialize_struct(&tb_meta)?), /* (tenant, db_id, tb_id) -> tb_meta */
@@ -1307,7 +1305,6 @@ impl<KV: KVApi> SchemaApi for KV {
                     if_then: vec![
                         // Changing a table in a db has to update the seq of db_meta,
                         // to block the batch-delete-tables when deleting a db.
-                        // TODO: test this when old metasrv is replaced with kv-txn based SchemaApi.
                         txn_op_put(&DatabaseId { db_id }, serialize_struct(&db_meta)?), /* (db_id) -> db_meta */
                         txn_op_put(&dbid_tbname, serialize_u64(table_id)?), /* (tenant, db_id, tb_name) -> tb_id */
                         // txn_op_put(&dbid_tbname_idlist, serialize_struct(&tb_id_list)?)?, // _fd_table_id_list/db_id/table_name -> tb_id_list
@@ -1498,7 +1495,6 @@ impl<KV: KVApi> SchemaApi for KV {
                     txn_op_put(&newdbid_newtbname, serialize_u64(table_id)?), /* (db_id, new_tb_name) -> tb_id */
                     // Changing a table in a db has to update the seq of db_meta,
                     // to block the batch-delete-tables when deleting a db.
-                    // TODO: test this when old metasrv is replaced with kv-txn based SchemaApi.
                     txn_op_put(&DatabaseId { db_id }, serialize_struct(&db_meta)?), /* (db_id) -> db_meta */
                     txn_op_put(&dbid_tbname_idlist, serialize_struct(&tb_id_list)?), /* _fd_table_id_list/db_id/old_table_name -> tb_id_list */
                     txn_op_put(&new_dbid_tbname_idlist, serialize_struct(&new_tb_id_list)?), /* _fd_table_id_list/db_id/new_table_name -> tb_id_list */
@@ -1507,7 +1503,6 @@ impl<KV: KVApi> SchemaApi for KV {
 
                 if db_id != new_db_id {
                     then_ops.push(
-                        // TODO: test this when old metasrv is replaced with kv-txn based SchemaApi.
                         txn_op_put(
                             &DatabaseId { db_id: new_db_id },
                             serialize_struct(&new_db_meta)?,

src/meta/raft-store/src/state_machine/sm.rs

@@ -95,11 +95,11 @@ const TREE_STATE_MACHINE: &str = "state_machine";
 
 /// StateMachine subscriber trait
 pub trait StateMachineSubscriber: Debug + Sync + Send {
-    fn kv_changed(&self, key: &str, prev: Option<SeqV>, current: Option<SeqV>);
+    fn kv_changed(&self, change: Change<Vec<u8>, String>);
 }
 
 /// The state machine of the `MemStore`.
-/// It includes user data and two raft-related informations:
+/// It includes user data and two raft-related information:
 /// `last_applied_logs` and `client_serial_responses` to achieve idempotence.
 #[derive(Debug)]
 pub struct StateMachine {
@@ -342,9 +342,7 @@ impl StateMachine {
         // Send queued change events to subscriber
         if let Some(subscriber) = &self.subscriber {
             for event in changes {
-                // TODO: use Change as the event data type.
-                // safe unwrap
-                subscriber.kv_changed(&event.ident.unwrap(), event.prev, event.result);
+                subscriber.kv_changed(event);
             }
         }
 
@@ -638,17 +636,15 @@ impl StateMachine {
         if let Some(kv_pairs) = kv_pairs {
             if let Some(kv_pairs) = kv_pairs.get(delete_by_prefix) {
                 for (key, _seq) in kv_pairs.iter() {
-                    // TODO: return StorageError
-                    let ret = Self::txn_upsert_kv(txn_tree, &UpsertKV::delete(key), log_time_ms);
+                    let (expired, prev, res) =
+                        Self::txn_upsert_kv(txn_tree, &UpsertKV::delete(key), log_time_ms)?;
 
-                    if let Ok(ret) = ret {
-                        count += 1;
+                    count += 1;
 
-                        if ret.0.is_some() {
-                            txn_tree.push_change(key, ret.0.clone(), None);
-                        }
-                        txn_tree.push_change(key, ret.1.clone(), ret.2);
+                    if expired.is_some() {
+                        txn_tree.push_change(key, expired, None);
                     }
+                    txn_tree.push_change(key, prev, res);
                 }
             }
         }

src/meta/service/src/watcher/watcher_manager.rs

@@ -23,8 +23,8 @@ use common_meta_types::protobuf::watch_request::FilterType;
 use common_meta_types::protobuf::Event;
 use common_meta_types::protobuf::WatchRequest;
 use common_meta_types::protobuf::WatchResponse;
+use common_meta_types::Change;
 use common_meta_types::PbSeqV;
-use common_meta_types::SeqV;
 use prost::Message;
 use tonic::Status;
 use tracing::info;
@@ -39,13 +39,6 @@ pub type WatcherStreamSender = Sender<Result<WatchResponse, Status>>;
 
 type CreateWatcherEvent = (WatchRequest, WatcherStreamSender);
 
-#[derive(Clone, Debug)]
-pub struct StateMachineKvData {
-    pub key: String,
-    pub prev: Option<SeqV>,
-    pub current: Option<SeqV>,
-}
-
 #[derive(Clone, Debug)]
 pub struct WatcherStateMachineSubscriber {
     event_tx: mpsc::UnboundedSender<WatcherEvent>,
@@ -54,7 +47,7 @@ pub struct WatcherStateMachineSubscriber {
 #[derive(Clone)]
 pub enum WatcherEvent {
     CreateWatcherEvent(CreateWatcherEvent),
-    StateMachineKvDataEvent(StateMachineKvData),
+    StateMachineKvDataEvent(Change<Vec<u8>, String>),
 }
 
 #[derive(Debug)]
@@ -106,8 +99,8 @@ impl WatcherManagerCore {
                     WatcherEvent::CreateWatcherEvent((req, tx)) => {
                         self.create_watcher_stream(req, tx).await;
                     }
-                    WatcherEvent::StateMachineKvDataEvent(kv) => {
-                        self.notify_event(kv).await;
+                    WatcherEvent::StateMachineKvDataEvent(kv_change) => {
+                        self.notify_event(kv_change).await;
                     }
                 }
             } else {
@@ -124,14 +117,15 @@ impl WatcherManagerCore {
         server_metrics::incr_watchers(-1);
     }
 
-    async fn notify_event(&mut self, kv: StateMachineKvData) {
-        let set = self.watcher_range_map.get_by_point(&kv.key);
+    async fn notify_event(&mut self, change: Change<Vec<u8>, String>) {
+        let k = change.ident.as_ref().unwrap();
+        let set = self.watcher_range_map.get_by_point(k);
         if set.is_empty() {
             return;
         }
 
-        let current = kv.current;
-        let prev = kv.prev;
+        let current = change.result;
+        let prev = change.prev;
 
         let is_delete_event = current.is_none();
         let mut remove_range_keys: Vec<RangeMapKey<String, WatcherId>> = vec![];
@@ -151,7 +145,7 @@ impl WatcherManagerCore {
             assert_eq!(stream.id, watcher_id);
             let resp = WatchResponse {
                 event: Some(Event {
-                    key: kv.key.clone(),
+                    key: k.to_string(),
                     current: current.clone().map(PbSeqV::from),
                     prev: prev.clone().map(PbSeqV::from),
                 }),
@@ -217,13 +211,9 @@ impl WatcherManagerCore {
 }
 
 impl StateMachineSubscriber for WatcherStateMachineSubscriber {
-    fn kv_changed(&self, key: &str, prev: Option<SeqV>, current: Option<SeqV>) {
+    fn kv_changed(&self, change: Change<Vec<u8>, String>) {
         let _ = self
             .event_tx
-            .send(WatcherEvent::StateMachineKvDataEvent(StateMachineKvData {
-                key: key.to_string(),
-                prev,
-                current,
-            }));
+            .send(WatcherEvent::StateMachineKvDataEvent(change));
     }
 }