fn backup<E: Engine>(
        &self,
        writer: &mut BackupWriter,
        engine: &E,
        concurrency_manager: ConcurrencyManager,
        backup_ts: TimeStamp,
        begin_ts: TimeStamp,
    ) -> Result<Statistics> {
        assert!(!self.is_raw_kv);

        let mut ctx = Context::default();
        ctx.set_region_id(self.region.get_id());
        ctx.set_region_epoch(self.region.get_region_epoch().to_owned());
        ctx.set_peer(self.leader.clone());

        // Update max_read_ts and check the in-memory lock table before getting the snapshot
        concurrency_manager.update_max_read_ts(backup_ts);
        concurrency_manager
            .read_range_check(
                self.start_key.as_ref(),
                self.end_key.as_ref(),
                |key, lock| {
                    Lock::check_ts_conflict(
                        Cow::Borrowed(lock),
                        &key,
                        backup_ts,
                        &Default::default(),
                    )
                },
            )
            .map_err(MvccError::from)
            .map_err(TxnError::from)?;

        let snapshot = match engine.snapshot(&ctx) {
            Ok(s) => s,
            Err(e) => {
                error!("backup snapshot failed"; "error" => ?e);
                return Err(e.into());
            }
        };
        let snap_store = SnapshotStore::new(
            snapshot,
            backup_ts,
            IsolationLevel::Si,
            false, /* fill_cache */
            Default::default(),
            false,
        );
        let start_key = self.start_key.clone();
        let end_key = self.end_key.clone();
        // Incremental backup needs to output delete records.
        let incremental = !begin_ts.is_zero();
        let mut scanner = snap_store
            .entry_scanner(start_key, end_key, begin_ts, incremental)
            .unwrap();

        let start = Instant::now();
        let mut batch = EntryBatch::with_capacity(BACKUP_BATCH_LIMIT);
        loop {
            if let Err(e) = scanner.scan_entries(&mut batch) {
                error!("backup scan entries failed"; "error" => ?e);
                return Err(e.into());
            };
            if batch.is_empty() {
                break;
            }
            debug!("backup scan entries"; "len" => batch.len());
            // Build sst files.
            if let Err(e) = writer.write(batch.drain(), true) {
                error!("backup build sst failed"; "error" => ?e);
                return Err(e);
            }
        }
        BACKUP_RANGE_HISTOGRAM_VEC
            .with_label_values(&["scan"])
            .observe(start.elapsed().as_secs_f64());
        let stat = scanner.take_statistics();
        Ok(stat)
    }
/// Get entries from the scanner and save them to storage
Get entries from the scanner and save them to storage
Get entries from the scanner and save them to storage
fn backup<E: Engine>(
        &self,
        writer: &mut BackupWriter,
        engine: &E,
        concurrency_manager: ConcurrencyManager,
        backup_ts: TimeStamp,
        begin_ts: TimeStamp,
    ) -> Result<Statistics> {
        assert!(!self.is_raw_kv);
        let mut ctx = Context::default();
        ctx.set_region_id(self.region.get_id());
        ctx.set_region_epoch(self.region.get_region_epoch().to_owned());
        ctx.set_peer(self.leader.clone());
        // Update max_read_ts and check the in-memory lock table before getting the snapshot
        concurrency_manager.update_max_read_ts(backup_ts);
        concurrency_manager
            .read_range_check(
                self.start_key.as_ref(),
                self.end_key.as_ref(),
                |key, lock| {
                    Lock::check_ts_conflict(
                        Cow::Borrowed(lock),
                        &key,
                        backup_ts,
                        &Default::default(),
                    )
                },
            )
            .map_err(MvccError::from)
            .map_err(TxnError::from)?;
        let snapshot = match engine.snapshot(&ctx) {
            Ok(s) => s,
            Err(e) => {
                error!("backup snapshot failed"; "error" => ?e);
                return Err(e.into());
            }
        };
        let snap_store = SnapshotStore::new(
            snapshot,
            backup_ts,
            IsolationLevel::Si,
            false, /* fill_cache */
            Default::default(),
            false,
        );
        let start_key = self.start_key.clone();
        let end_key = self.end_key.clone();
        // Incremental backup needs to output delete records.
        let incremental = !begin_ts.is_zero();
        let mut scanner = snap_store
            .entry_scanner(start_key, end_key, begin_ts, incremental)
            .unwrap();
        let start = Instant::now();
        let mut batch = EntryBatch::with_capacity(BACKUP_BATCH_LIMIT);
        loop {
            if let Err(e) = scanner.scan_entries(&mut batch) {
                error!("backup scan entries failed"; "error" => ?e);
                return Err(e.into());
            };
            if batch.is_empty() {
                break;
            }
            debug!("backup scan entries"; "len" => batch.len());
            // Build sst files.
            if let Err(e) = writer.write(batch.drain(), true) {
                error!("backup build sst failed"; "error" => ?e);
                return Err(e);
            }
        }
        BACKUP_RANGE_HISTOGRAM_VEC
            .with_label_values(&["scan"])
            .observe(start.elapsed().as_secs_f64());
        let stat = scanner.take_statistics();
        Ok(stat)
    }
fn backup < E : Engine > ( & self , writer : & mut BackupWriter , engine : & E , concurrency_manager : ConcurrencyManager , backup_ts : TimeStamp , begin_ts : TimeStamp , ) -> Result < Statistics > { assert ! ( self is_raw_kv ) ; let mut ctx = Context :: default ( ) ; ctx . set_region_id ( self . region . get_id ( ) ) ; ctx . set_region_epoch ( self . region . get_region_epoch ( ) . to_owned ( ) ) ; ctx . set_peer ( self . leader . clone ( ) ) ; // Update max_read_ts and check the in-memory lock table before getting the snapshot concurrency_manager . update_max_read_ts ( backup_ts ) ; concurrency_manager . read_range_check ( self . start_key . as_ref ( ) , self . end_key . as_ref ( ) , | key , lock | { Lock :: check_ts_conflict ( Cow :: Borrowed ( lock ) , & key , backup_ts , & Default :: default ( ) , ) } , ) . map_err ( MvccError :: from ) . map_err ( TxnError :: from ) ? ; let snapshot = match engine . snapshot ( & ctx ) { Ok ( s ) => s , Err ( e ) => { error ! ( " " " " e ) ; return Err ( e . into ( ) ) ; } } ; let snap_store = SnapshotStore :: new ( snapshot , backup_ts , IsolationLevel :: Si , false , /* fill_cache */ Default :: default ( ) , false , ) ; let start_key = self . start_key . clone ( ) ; let end_key = self . end_key . clone ( ) ; // Incremental backup needs to output delete records. let incremental = ! begin_ts . is_zero ( ) ; let mut scanner = snap_store . entry_scanner ( start_key , end_key , begin_ts , incremental ) . unwrap ( ) ; let start = Instant :: now ( ) ; let mut batch = EntryBatch :: with_capacity ( BACKUP_BATCH_LIMIT ) ; loop { if let Err ( e ) = scanner . scan_entries ( & mut batch ) { error ! ( " " " " e ) ; return Err ( e . into ( ) ) ; } ; if batch . is_empty ( ) { break ; } debug ! ( " " " " batch len ( ) ) ; // Build sst files. if let Err ( e ) = writer . write ( batch . drain ( ) , true ) { error ! ( " " " " e ) ; return Err ( e ) ; } } BACKUP_RANGE_HISTOGRAM_VEC . with_label_values ( & [ " " ] ) . observe ( start . elapsed ( ) . as_secs_f64 ( ) ) ; let stat = scanner . take_statistics ( ) ; Ok ( stat ) }
Get entries from the scanner and save them to storage
Get entries from the scanner and save them to storage
&self
writer
&mut BackupWriter
engine
&E
concurrency_manager
ConcurrencyManager
backup_ts
TimeStamp
begin_ts
TimeStamp