if !self.fsm.peer.is_leader() { 
    return; 
} 
 
if self.fsm.peer.may_skip_split_check 
    && self.fsm.peer.compaction_declined_bytes < self.ctx.cfg.region_split_check_diff().0 
    && self.fsm.peer.size_diff_hint < self.ctx.cfg.region_split_check_diff().0 
{ 
    return; 
} 

>> operator add split-region 1 --policy=approximate     // 将 Region 1 对半拆分成两个 Region，基于粗略估计值 
>> operator add split-region 1 --policy=scan            // 将 Region 1 对半拆分成两个 Region，基于精确扫描值 

if resp.has_split_region() { 
    let mut split_region = resp.take_split_region(); 
    info!("try to split"; "region_id" => region_id, "region_epoch" => ?epoch); 
    let msg = if split_region.get_policy() == pdpb::CheckPolicy::Usekey { 
        CasualMessage::SplitRegion { 
            region_epoch: epoch, 
            split_keys: split_region.take_keys().into(), 
            callback: Callback::None, 
            source: "pd".into(), 
        } 
    } else { 
        CasualMessage::HalfSplitRegion { 
            region_epoch: epoch, 
            policy: split_region.get_policy(), 
            source: "pd", 
            cb: Callback::None, 
        } 
    }; 
    if let Err(e) = router.send(region_id, PeerMsg::CasualMessage(msg)) { 
        error!("send halfsplit request failed"; "region_id" => region_id, "err" => ?e); 
    } 
} 

/// SplitChecker is invoked during a split check scan, and decides to use 
/// which keys to split a region. 
pub trait SplitChecker<E> { 
    /// Hook to call for every kv scanned during split. 
    /// 
    /// Return true to abort scan early. 
    fn on_kv(&mut self, _: &mut ObserverContext<'_>, _: &KeyEntry) -> bool { 
        false 
    } 
 
    /// Get the desired split keys. 
    fn split_keys(&mut self) -> Vec<Vec<u8>>; 
 
    /// Get approximate split keys without scan. 
    fn approximate_split_keys(&mut self, _: &Region, _: &E) -> Result<Vec<Vec<u8>>> { 
        Ok(vec![]) 
    } 
 
    /// Get split policy. 
    fn policy(&self) -> CheckPolicy; 
} 

fn half_split_bucket_size(region_max_size: u64) -> u64 { 
    let mut half_split_bucket_size = region_max_size / BUCKET_NUMBER_LIMIT as u64; 
    let bucket_size_limit = ReadableSize::mb(BUCKET_SIZE_LIMIT_MB).0; 
    if half_split_bucket_size == 0 { 
        half_split_bucket_size = 1; 
    } else if half_split_bucket_size > bucket_size_limit { 
        half_split_bucket_size = bucket_size_limit; 
    } 
    half_split_bucket_size 
} 

fn on_kv(&mut self, _: &mut ObserverContext<'_>, entry: &KeyEntry) -> bool { 
    if self.buckets.is_empty() || self.cur_bucket_size >= self.each_bucket_size { 
        self.buckets.push(entry.key().to_vec()); 
        self.cur_bucket_size = 0; 
    } 
    self.cur_bucket_size += entry.entry_size() as u64; 
    false 
} 
 
fn split_keys(&mut self) -> Vec<Vec<u8>> { 
    let mid = self.buckets.len() / 2; 
    if mid == 0 { 
        vec![] 
    } else { 
        let data_key = self.buckets.swap_remove(mid); 
        let key = keys::origin_key(&data_key).to_vec(); 
        vec![key] 
    } 
} 

fn on_kv(&mut self, _: &mut ObserverContext<'_>, entry: &KeyEntry) -> bool { 
    let size = entry.entry_size() as u64; 
    self.current_size += size; 
 
    let mut over_limit = self.split_keys.len() as u64 >= self.batch_split_limit; 
    if self.current_size > self.split_size && !over_limit { 
        self.split_keys.push(keys::origin_key(entry.key()).to_vec()); 
        // if for previous on_kv() self.current_size == self.split_size, 
        // the split key would be pushed this time, but the entry size for this time should not be ignored. 
        self.current_size = if self.current_size - size == self.split_size { 
            size 
        } else { 
            0 
        }; 
        over_limit = self.split_keys.len() as u64 >= self.batch_split_limit; 
    } 
 
    // For a large region, scan over the range maybe cost too much time, 
    // so limit the number of produced split_key for one batch. 
    // Also need to scan over self.max_size for last part. 
    over_limit && self.current_size + self.split_size >= self.max_size 
} 
 
fn split_keys(&mut self) -> Vec<Vec<u8>> { 
    // make sure not to split when less than max_size for last part 
    if self.current_size + self.split_size < self.max_size { 
        self.split_keys.pop(); 
    } 
    if !self.split_keys.is_empty() { 
        std::mem::take(&mut self.split_keys) 
    } else { 
        vec![] 
    } 
} 

/// Feed keys in order to find the split key. 
/// If `current_data_key` does not belong to `status.first_encoded_table_prefix`. 
/// it returns the encoded table prefix of `current_data_key`. 
fn on_kv(&mut self, _: &mut ObserverContext<'_>, entry: &KeyEntry) -> bool { 
    if self.split_key.is_some() { 
        return true; 
    } 
 
    let current_encoded_key = keys::origin_key(entry.key()); 
 
    let split_key = if self.first_encoded_table_prefix.is_some() { 
        if !is_same_table( 
            self.first_encoded_table_prefix.as_ref().unwrap(), 
            current_encoded_key, 
        ) { 
            // Different tables. 
            Some(current_encoded_key) 
        } else { 
            None 
        } 
    } else if is_table_key(current_encoded_key) { 
        // Now we meet the very first table key of this region. 
        Some(current_encoded_key) 
    } else { 
        None 
    }; 
    self.split_key = split_key.and_then(to_encoded_table_prefix); 
    self.split_key.is_some() 
} 

pub fn new<C: CasualRouter<E> + Clone + Send + 'static>( 
    ch: C, 
    cfg: Config, 
) -> CoprocessorHost<E> { 
    let mut registry = Registry::default(); 
    registry.register_split_check_observer( 
        200, 
        BoxSplitCheckObserver::new(SizeCheckObserver::new(ch.clone())), 
    ); 
    registry.register_split_check_observer( 
        200, 
        BoxSplitCheckObserver::new(KeysCheckObserver::new(ch)), 
    ); 
    registry.register_split_check_observer(100, BoxSplitCheckObserver::new(HalfCheckObserver)); 
    registry.register_split_check_observer( 
        400, 
        BoxSplitCheckObserver::new(TableCheckObserver::default()), 
    ); 
    CoprocessorHost { registry, cfg } 
} 

#[derive(Clone)] 
pub struct HalfCheckObserver; 
 
impl Coprocessor for HalfCheckObserver {} 
 
impl<E> SplitCheckObserver<E> for HalfCheckObserver 
where 
    E: KvEngine, 
{ 
    fn add_checker( 
        &self, 
        _: &mut ObserverContext<'_>, 
        host: &mut Host<'_, E>, 
        _: &E, 
        policy: CheckPolicy, 
    ) { 
        if host.auto_split() { 
            return; 
        } 
        host.add_checker(Box::new(Checker::new( 
            half_split_bucket_size(host.cfg.region_max_size().0), 
            policy, 
        ))) 
    } 
} 

#[derive(Default, Clone)] 
pub struct TableCheckObserver; 
 
impl Coprocessor for TableCheckObserver {} 
 
impl<E> SplitCheckObserver<E> for TableCheckObserver 
where 
    E: KvEngine, 
{ 
    fn add_checker( 
        &self, 
        ctx: &mut ObserverContext<'_>, 
        host: &mut Host<'_, E>, 
        engine: &E, 
        policy: CheckPolicy, 
    ) { 
        if !host.cfg.split_region_on_table { 
            return; 
        } 
        ... 
} 

info!( 
    "try to batch split region"; 
    "region_id" => region.get_id(), 
    "new_region_ids" => ?resp.get_ids(), 
    "region" => ?region, 
    "task" => task, 
); 
 
let req = new_batch_split_region_request( 
    split_keys, 
    resp.take_ids().into(), 
    right_derive, 
); 
let region_id = region.get_id(); 
let epoch = region.take_region_epoch(); 
send_admin_request( 
    &router, 
    region_id, 
    epoch, 
    peer, 
    req, 
    callback, 
    Default::default(), 
); 

for new_region in &regions { 
    if new_region.get_id() == derived.get_id() { 
        continue; 
    } 
    let new_split_peer = new_split_regions.get(&new_region.get_id()).unwrap(); 
    if let Some(ref r) = new_split_peer.result { 
        warn!( 
            "new region from splitting already exists"; 
            "new_region_id" => new_region.get_id(), 
            "new_peer_id" => new_split_peer.peer_id, 
            "reason" => r, 
            "region_id" => self.region_id(), 
            "peer_id" => self.id(), 
        ); 
        continue; 
    } 
    write_peer_state(kv_wb_mut, new_region, PeerState::Normal, None) 
        .and_then(|_| write_initial_apply_state(kv_wb_mut, new_region.get_id())) 
        .unwrap_or_else(|e| { 
            panic!( 
                "{} fails to save split region {:?}: {:?}", 
                self.tag, new_region, e 
            ) 
        }); 
} 
write_peer_state(kv_wb_mut, &derived, PeerState::Normal, None).unwrap_or_else(|e| { 
    panic!("{} fails to update region {:?}: {:?}", self.tag, derived, e) 
}); 

if is_leader { 
    self.fsm.peer.approximate_size = estimated_size; 
    self.fsm.peer.approximate_keys = estimated_keys; 
    self.fsm.peer.heartbeat_pd(self.ctx); 
    // Notify pd immediately to let it update the region meta. 
    info!( 
        "notify pd with split"; 
        "region_id" => self.fsm.region_id(), 
        "peer_id" => self.fsm.peer_id(), 
        "split_count" => regions.len(), 
    ); 
    // Now pd only uses ReportBatchSplit for history operation show, 
    // so we send it independently here. 
    let task = PdTask::ReportBatchSplit { 
        regions: regions.to_vec(), 
    }; 
    if let Err(e) = self.ctx.pd_scheduler.schedule(task) { 
        error!( 
            "failed to notify pd"; 
            "region_id" => self.fsm.region_id(), 
            "peer_id" => self.fsm.peer_id(), 
            "err" => %e, 
        ); 
    } 
}