Added a new observer type 'pitr'

components/backup-stream/src/event_loader.rs

@@ -346,7 +346,7 @@ where
                     .register_region(&r.region, handle.clone(), Some(start_ts));
                 let region_id = r.region.get_id();
                 let snap = self.observe_over_with_retry(&r.region, move || {
-                    ChangeObserver::from_cdc(region_id, handle.clone())
+                    ChangeObserver::from_pitr(region_id, handle.clone())
                 })?;
                 let stat = self.do_initial_scan(&r.region, start_ts, snap)?;
                 total_stat.add_statistics(&stat);

components/backup-stream/src/observer.rs

@@ -252,7 +252,8 @@ mod tests {
         }
 
         // Test events with key in the range can be observed.
-        let observe_info = CmdObserveInfo::from_handle(handle.clone(), ObserveHandle::new());
+        let observe_info =
+            CmdObserveInfo::from_handle(handle.clone(), ObserveHandle::new(), ObserveHandle::new());
         let mut cb = CmdBatch::new(&observe_info, 42);
         cb.push(&observe_info, 42, Cmd::default());
         let mut cmd_batches = vec![cb];
@@ -263,7 +264,11 @@ mod tests {
         );
 
         // Test event from other region should not be send.
-        let observe_info = CmdObserveInfo::from_handle(ObserveHandle::new(), ObserveHandle::new());
+        let observe_info = CmdObserveInfo::from_handle(
+            ObserveHandle::new(),
+            ObserveHandle::new(),
+            ObserveHandle::new(),
+        );
         let mut cb = CmdBatch::new(&observe_info, 43);
         cb.push(&observe_info, 43, Cmd::default());
         cb.level = ObserveLevel::None;

components/cdc/src/observer.rs

@@ -207,7 +207,11 @@ mod tests {
     fn test_register_and_deregister() {
         let (scheduler, mut rx) = tikv_util::worker::dummy_scheduler();
         let observer = CdcObserver::new(scheduler);
-        let observe_info = CmdObserveInfo::from_handle(ObserveHandle::new(), ObserveHandle::new());
+        let observe_info = CmdObserveInfo::from_handle(
+            ObserveHandle::new(),
+            ObserveHandle::new(),
+            ObserveHandle::new(),
+        );
         let engine = TestEngineBuilder::new().build().unwrap().get_rocksdb();
 
         let mut cb = CmdBatch::new(&observe_info, 0);
@@ -228,6 +232,7 @@ mod tests {
 
         // Stop observing cmd
         observe_info.cdc_id.stop_observing();
+        observe_info.pitr_id.stop_observing();
         let mut cb = CmdBatch::new(&observe_info, 0);
         cb.push(&observe_info, 0, Cmd::default());
         <CdcObserver as CmdObserver<RocksEngine>>::on_flush_applied_cmd_batch(
@@ -238,7 +243,7 @@ mod tests {
         );
         match rx.recv_timeout(Duration::from_millis(10)) {
             Err(std::sync::mpsc::RecvTimeoutError::Timeout) => {}
-            _ => panic!("unexpected result"),
+            any => panic!("unexpected result: {:?}", any),
         };
 
         // Does not send unsubscribed region events.

components/raftstore/src/coprocessor/dispatcher.rs

@@ -737,7 +737,11 @@ mod tests {
         host.post_apply_sst_from_snapshot(&region, "default", "");
         assert_all!([&ob.called], &[55]);
 
-        let observe_info = CmdObserveInfo::from_handle(ObserveHandle::new(), ObserveHandle::new());
+        let observe_info = CmdObserveInfo::from_handle(
+            ObserveHandle::new(),
+            ObserveHandle::new(),
+            ObserveHandle::new(),
+        );
         let mut cb = CmdBatch::new(&observe_info, 0);
         cb.push(&observe_info, 0, Cmd::default());
         host.on_flush_applied_cmd_batch(cb.level, vec![cb], &PanicEngine);

components/raftstore/src/coprocessor/mod.rs

@@ -1,6 +1,5 @@
 // Copyright 2016 TiKV Project Authors. Licensed under Apache-2.0.
 
-use std::cmp;
 use std::fmt::{self, Debug, Formatter};
 use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
 use std::sync::Arc;
@@ -247,11 +246,20 @@ impl ObserveHandle {
 pub struct CmdObserveInfo {
     pub cdc_id: ObserveHandle,
     pub rts_id: ObserveHandle,
+    pub pitr_id: ObserveHandle,
 }
 
 impl CmdObserveInfo {
-    pub fn from_handle(cdc_id: ObserveHandle, rts_id: ObserveHandle) -> CmdObserveInfo {
-        CmdObserveInfo { cdc_id, rts_id }
+    pub fn from_handle(
+        cdc_id: ObserveHandle,
+        rts_id: ObserveHandle,
+        pitr_id: ObserveHandle,
+    ) -> CmdObserveInfo {
+        CmdObserveInfo {
+            cdc_id,
+            rts_id,
+            pitr_id,
+        }
     }
 
     fn observe_level(&self) -> ObserveLevel {
@@ -261,13 +269,18 @@ impl CmdObserveInfo {
         } else {
             ObserveLevel::None
         };
+        let pitr = if self.pitr_id.is_observing() {
+            ObserveLevel::All
+        } else {
+            ObserveLevel::None
+        };
         let rts = if self.rts_id.is_observing() {
             // `resolved-ts` observe lock related data
             ObserveLevel::LockRelated
         } else {
             ObserveLevel::None
         };
-        cmp::max(cdc, rts)
+        cdc.max(rts).max(pitr)
     }
 }
 
@@ -276,6 +289,7 @@ impl Debug for CmdObserveInfo {
         f.debug_struct("CmdObserveInfo")
             .field("cdc_id", &self.cdc_id.id)
             .field("rts_id", &self.rts_id.id)
+            .field("pitr_id", &self.pitr_id.id)
             .finish()
     }
 }
@@ -296,6 +310,7 @@ pub struct CmdBatch {
     pub level: ObserveLevel,
     pub cdc_id: ObserveID,
     pub rts_id: ObserveID,
+    pub pitr_id: ObserveID,
     pub region_id: u64,
     pub cmds: Vec<Cmd>,
 }
@@ -306,6 +321,7 @@ impl CmdBatch {
             level: observe_info.observe_level(),
             cdc_id: observe_info.cdc_id.id,
             rts_id: observe_info.rts_id.id,
+            pitr_id: observe_info.pitr_id.id,
             region_id,
             cmds: Vec::new(),
         }
@@ -315,6 +331,7 @@ impl CmdBatch {
         assert_eq!(region_id, self.region_id);
         assert_eq!(observe_info.cdc_id.id, self.cdc_id);
         assert_eq!(observe_info.rts_id.id, self.rts_id);
+        assert_eq!(observe_info.pitr_id.id, self.pitr_id);
         self.cmds.push(cmd)
     }
 
@@ -376,22 +393,47 @@ mod tests {
     #[test]
     fn test_observe_level() {
         // Both cdc and `resolved-ts` are observing
-        let observe_info = CmdObserveInfo::from_handle(ObserveHandle::new(), ObserveHandle::new());
+        let observe_info = CmdObserveInfo::from_handle(
+            ObserveHandle::new(),
+            ObserveHandle::new(),
+            ObserveHandle::new(),
+        );
         assert_eq!(observe_info.observe_level(), ObserveLevel::All);
 
         // No observer
         observe_info.cdc_id.stop_observing();
         observe_info.rts_id.stop_observing();
+        observe_info.pitr_id.stop_observing();
         assert_eq!(observe_info.observe_level(), ObserveLevel::None);
 
         // Only cdc observing
-        let observe_info = CmdObserveInfo::from_handle(ObserveHandle::new(), ObserveHandle::new());
+        let observe_info = CmdObserveInfo::from_handle(
+            ObserveHandle::new(),
+            ObserveHandle::new(),
+            ObserveHandle::new(),
+        );
         observe_info.rts_id.stop_observing();
+        observe_info.pitr_id.stop_observing();
         assert_eq!(observe_info.observe_level(), ObserveLevel::All);
 
         // Only `resolved-ts` observing
-        let observe_info = CmdObserveInfo::from_handle(ObserveHandle::new(), ObserveHandle::new());
+        let observe_info = CmdObserveInfo::from_handle(
+            ObserveHandle::new(),
+            ObserveHandle::new(),
+            ObserveHandle::new(),
+        );
         observe_info.cdc_id.stop_observing();
+        observe_info.pitr_id.stop_observing();
         assert_eq!(observe_info.observe_level(), ObserveLevel::LockRelated);
+
+        // Only `backup-stream(pitr)` observing
+        let observe_info = CmdObserveInfo::from_handle(
+            ObserveHandle::new(),
+            ObserveHandle::new(),
+            ObserveHandle::new(),
+        );
+        observe_info.cdc_id.stop_observing();
+        observe_info.rts_id.stop_observing();
+        assert_eq!(observe_info.observe_level(), ObserveLevel::All);
     }
 }