Repository: matrixorigin/matrixcube
Pull request #707 (2022-03-14 06:08:31)

refactor component/prophet/core & component/prophet/storage
**What type of PR is this?**

- [ ] API-change
- [ ] BUG
- [ ] Improvement
- [ ] Documentation
- [ ] Feature
- [ ] Test and CI
- [x] Code Refactoring

**Which issue(s) this PR fixes:**

Fixes #

**What this PR does / why we need it:**

Not Available

**Special notes for your reviewer:**

Not Available

**Additional documentation (e.g. design docs, usage docs, etc.):**

Not Available



Patch:
diff --git a/components/prophet/cluster/cluster.go b/components/prophet/cluster/cluster.go
index 66e4f68..7b7ebf4 100644
--- a/components/prophet/cluster/cluster.go
+++ b/components/prophet/cluster/cluster.go
@@ -51,7 +51,7 @@ const (
 	clientTimeout            = 3 * time.Second
 	defaultChangedEventLimit = 10000
 	// persistLimitRetryTimes is used to reduce the probability of the persistent error
-	// since the once the store is add or remove, we shouldn't return an error even if the store limit is failed to persist.
+	// since the once the store is added or removed, we shouldn't return an error even if the store limit is failed to persist.
 	persistLimitRetryTimes = 5
 	persistLimitWaitTime   = 100 * time.Millisecond
 )
@@ -573,7 +573,7 @@ func (c *RaftCluster) processShardHeartbeat(res *core.CachedShard) error {
 	}
 
 	c.Lock()
-	inCreating := c.core.IsWaittingCreateShard(res.Meta.GetID())
+	inCreating := c.core.IsWaitingCreateShard(res.Meta.GetID())
 	if isNew && inCreating {
 		if c.resourceStateChangedHandler != nil {
 			c.resourceStateChangedHandler(&res.Meta, metapb.ShardState_Creating,
@@ -700,11 +700,6 @@ func (c *RaftCluster) GetShardByKey(group uint64, resourceKey []byte) *core.Cach
 	return c.core.SearchShard(group, resourceKey)
 }
 
-// GetPrevShardByKey gets previous resource and leader peer by the resource key from cluster.
-func (c *RaftCluster) GetPrevShardByKey(group uint64, resourceKey []byte) *core.CachedShard {
-	return c.core.SearchPrevShard(group, resourceKey)
-}
-
 // ScanShards scans resource with start key, until the resource contains endKey, or
 // total number greater than limit.
 func (c *RaftCluster) ScanShards(group uint64, startKey, endKey []byte, limit int) []*core.CachedShard {
@@ -722,7 +717,7 @@ func (c *RaftCluster) GetShard(resourceID uint64) *core.CachedShard {
 }
 
 // GetMetaShards gets resources from cluster.
-func (c *RaftCluster) GetMetaShards() []*metapb.Shard {
+func (c *RaftCluster) GetMetaShards() []metapb.Shard {
 	return c.core.GetMetaShards()
 }
 
@@ -1283,7 +1278,7 @@ func (c *RaftCluster) updateShardsLabelLevelStats(resources []*core.CachedShard)
 func (c *RaftCluster) takeShardStoresLocked(res *core.CachedShard) []*core.CachedStore {
 	containers := make([]*core.CachedStore, 0, len(res.Meta.GetReplicas()))
 	for _, p := range res.Meta.GetReplicas() {
-		if container := c.core.TakeStore(p.StoreID); container != nil {
+		if container := c.core.GetStore(p.StoreID); container != nil {
 			containers = append(containers, container)
 		}
 	}
diff --git a/components/prophet/cluster/cluster_worker.go b/components/prophet/cluster/cluster_worker.go
index aab4c18..bda87f8 100644
--- a/components/prophet/cluster/cluster_worker.go
+++ b/components/prophet/cluster/cluster_worker.go
@@ -238,7 +238,7 @@ func (c *RaftCluster) HandleCreateShards(request *rpcpb.ProphetRequest) (*rpcpb.
 			}
 		}
 		if create {
-			c.core.ForeachWaittingCreateShards(func(wres metapb.Shard) {
+			c.core.ForeachWaitingCreateShards(func(wres metapb.Shard) {
 				if wres.GetUnique() == res.GetUnique() {
 					create = false
 					c.logger.Info("resource already in waitting create queue",
@@ -296,7 +296,7 @@ func (c *RaftCluster) HandleCreateShards(request *rpcpb.ProphetRequest) (*rpcpb.
 		return nil, err
 	}
 
-	c.core.AddWaittingCreateShards(createdShards...)
+	c.core.AddWaitingCreateShards(createdShards...)
 	c.triggerNotifyCreateShards()
 	return &rpcpb.CreateShardsRsp{}, nil
 }
@@ -401,7 +401,7 @@ func (c *RaftCluster) triggerNotifyCreateShards() {
 }
 
 func (c *RaftCluster) doNotifyCreateShards() {
-	c.core.ForeachWaittingCreateShards(func(res metapb.Shard) {
+	c.core.ForeachWaitingCreateShards(func(res metapb.Shard) {
 		c.addNotifyLocked(event.NewShardEvent(res, 0, false, true))
 	})
 }
diff --git a/components/prophet/config/config_job.go b/components/prophet/config/config_job.go
index e74f2bc..5885285 100644
--- a/components/prophet/config/config_job.go
+++ b/components/prophet/config/config_job.go
@@ -16,7 +16,7 @@ type jobRegister struct {
 // ShardsAware resources aware
 type ShardsAware interface {
 	// ForeachWaittingCreateShards do every waitting resources
-	ForeachWaittingCreateShards(do func(res metapb.Shard))
+	ForeachWaitingCreateShards(do func(res metapb.Shard))
 	// ForeachShards foreach resource by group
 	ForeachShards(group uint64, fn func(res metapb.Shard))
 	// GetShard returns resource runtime info
diff --git a/components/prophet/core/basic_cluster.go b/components/prophet/core/basic_cluster.go
index 1799da0..57da9fc 100644
--- a/components/prophet/core/basic_cluster.go
+++ b/components/prophet/core/basic_cluster.go
@@ -32,14 +32,14 @@ import (
 // BasicCluster provides basic data member and interface for a storage application cluster.
 type BasicCluster struct {
 	sync.RWMutex
-	logger               *zap.Logger
-	Stores               *CachedStores
-	Shards               *CachedShards
-	DestroyedShards      *roaring64.Bitmap
-	WaittingCreateShards map[uint64]metapb.Shard
-	DestroyingStatuses   map[uint64]*metapb.DestroyingStatus
-	ScheduleGroupRules   []metapb.ScheduleGroupRule
-	ScheduleGroupKeys    map[string]struct{}
+	logger              *zap.Logger
+	Stores              *StoresContainer
+	Shards              *ShardsContainer
+	DestroyedShards     *roaring64.Bitmap
+	WaitingCreateShards map[uint64]metapb.Shard
+	DestroyingStatuses  map[uint64]*metapb.DestroyingStatus
+	ScheduleGroupRules  []metapb.ScheduleGroupRule
+	ScheduleGroupKeys   map[string]struct{}
 }
 
 // NewBasicCluster creates a BasicCluster.
@@ -61,11 +61,11 @@ func (bc *BasicCluster) Reset() {
 	bc.Stores = NewCachedStores()
 	bc.Shards = NewCachedShards()
 	bc.DestroyedShards = roaring64.NewBitmap()
-	bc.WaittingCreateShards = make(map[uint64]metapb.Shard)
+	bc.WaitingCreateShards = make(map[uint64]metapb.Shard)
 	bc.ScheduleGroupRules = bc.ScheduleGroupRules[:0]
 }
 
-// AddRemovedShards add removed resources
+// AddRemovedShards add removed shards
 func (bc *BasicCluster) AddRemovedShards(ids ...uint64) {
 	bc.Lock()
 	defer bc.Unlock()
@@ -78,25 +78,25 @@ func (bc *BasicCluster) AddRemovedShards(ids ...uint64) {
 	}
 }
 
-// AddWaittingCreateShards add waitting create resources
-func (bc *BasicCluster) AddWaittingCreateShards(resources ...metapb.Shard) {
+// AddWaitingCreateShards add waiting create shards
+func (bc *BasicCluster) AddWaitingCreateShards(resources ...metapb.Shard) {
 	bc.Lock()
 	defer bc.Unlock()
 	for _, res := range resources {
-		bc.WaittingCreateShards[res.GetID()] = res
+		bc.WaitingCreateShards[res.GetID()] = res
 	}
 }
 
-// ForeachWaittingCreateShards do func for every waitting create resources
-func (bc *BasicCluster) ForeachWaittingCreateShards(fn func(res metapb.Shard)) {
+// ForeachWaitingCreateShards do func for every waiting create shards
+func (bc *BasicCluster) ForeachWaitingCreateShards(fn func(res metapb.Shard)) {
 	bc.RLock()
 	defer bc.RUnlock()
-	for _, res := range bc.WaittingCreateShards {
+	for _, res := range bc.WaitingCreateShards {
 		fn(res)
 	}
 }
 
-// ForeachShards foreach resources by group
+// ForeachShards foreach shards by group
 func (bc *BasicCluster) ForeachShards(group uint64, fn func(res metapb.Shard)) {
 	bc.RLock()
 	defer bc.RUnlock()
@@ -104,12 +104,12 @@ func (bc *BasicCluster) ForeachShards(group uint64, fn func(res metapb.Shard)) {
 	bc.Shards.ForeachShards(group, fn)
 }
 
-// IsWaittingCreateShard returns true means the resource is waitting create
-func (bc *BasicCluster) IsWaittingCreateShard(id uint64) bool {
+// IsWaitingCreateShard returns true means the resource is waiting create
+func (bc *BasicCluster) IsWaitingCreateShard(id uint64) bool {
 	bc.RLock()
 	defer bc.RUnlock()
 
-	_, ok := bc.WaittingCreateShards[id]
+	_, ok := bc.WaitingCreateShards[id]
 	return ok
 }
 
@@ -121,18 +121,18 @@ func (bc *BasicCluster) AlreadyRemoved(id uint64) bool {
 	return bc.DestroyedShards.Contains(id)
 }
 
-// GetDestroyShards get destroyed and destroying state resources
+// GetDestroyShards get destroyed and destroying state shards
 func (bc *BasicCluster) GetDestroyShards(bm *roaring64.Bitmap) (*roaring64.Bitmap, *roaring64.Bitmap) {
 	bc.RLock()
 	defer bc.RUnlock()
 
-	// read destroyed resources
+	// read destroyed shards
 	destroyed := bc.DestroyedShards.Clone()
 	destroyed.And(bm)
 
-	// read destroying resources
+	// read destroying shards
 	destroying := roaring64.New()
-	for id, res := range bc.Shards.resources.m {
+	for id, res := range bc.Shards.shards.m {
 		if res.Meta.GetState() == metapb.ShardState_Destroying && bm.Contains(id) {
 			destroying.Add(id)
 		}
@@ -168,15 +168,15 @@ func (bc *BasicCluster) GetShard(resourceID uint64) *CachedShard {
 	return bc.Shards.GetShard(resourceID)
 }
 
-// GetShards gets all CachedShard from resourceMap.
+// GetShards gets all CachedShard from shardMap.
 func (bc *BasicCluster) GetShards() []*CachedShard {
 	bc.RLock()
 	defer bc.RUnlock()
 	return bc.Shards.GetShards()
 }
 
-// GetMetaShards gets a set of *metapb.Shard from resourceMap.
-func (bc *BasicCluster) GetMetaShards() []*metapb.Shard {
+// GetMetaShards gets a set of *metapb.Shard from shardMap.
+func (bc *BasicCluster) GetMetaShards() []metapb.Shard {
 	bc.RLock()
 	defer bc.RUnlock()
 	return bc.Shards.GetMetaShards()
@@ -269,7 +269,7 @@ func (bc *BasicCluster) GetScheduleGroupKeysWithPrefix(prefix string) []string {
 	var keys []string
 	for k := range bc.ScheduleGroupKeys {
 		if strings.HasPrefix(k, prefix) ||
-			(k == "" && prefix == string(util.EncodeGroupKey(0, nil, nil))) {
+			(k == "" && prefix == util.EncodeGroupKey(0, nil, nil)) {
 			keys = append(keys, k)
 		}
 	}
@@ -330,14 +330,14 @@ func (bc *BasicCluster) selectShard(resources []*CachedShard, opts ...ShardOptio
 	return nil
 }
 
-// GetShardCount gets the total count of CachedShard of resourceMap.
+// GetShardCount gets the total count of CachedShard of shardMap.
 func (bc *BasicCluster) GetShardCount() int {
 	bc.RLock()
 	defer bc.RUnlock()
 	return bc.Shards.GetShardCount()
 }
 
-// GetStoreCount returns the total count of CachedStores.
+// GetStoreCount returns the total count of StoresContainer.
 func (bc *BasicCluster) GetStoreCount() int {
 	bc.RLock()
 	defer bc.RUnlock()
@@ -374,14 +374,14 @@ func (bc *BasicCluster) GetStorePendingPeerCount(groupKey string, containerID ui
 	return bc.Shards.GetStorePendingPeerCount(groupKey, containerID)
 }
 
-// GetStoreLeaderShardSize get total size of container's leader resources.
+// GetStoreLeaderShardSize get total size of container's leader shards.
 func (bc *BasicCluster) GetStoreLeaderShardSize(groupKey string, containerID uint64) int64 {
 	bc.RLock()
 	defer bc.RUnlock()
 	return bc.Shards.GetStoreLeaderShardSize(groupKey, containerID)
 }
 
-// GetStoreShardSize get total size of container's resources.
+// GetStoreShardSize get total size of container's shards.
 func (bc *BasicCluster) GetStoreShardSize(groupKey string, containerID uint64) int64 {
 	bc.RLock()
 	defer bc.RUnlock()
@@ -411,13 +411,6 @@ func (bc *BasicCluster) DeleteStore(container *CachedStore) {
 	bc.Stores.DeleteStore(container)
 }
 
-// TakeStore returns the point of the origin CachedStores with the specified containerID.
-func (bc *BasicCluster) TakeStore(containerID uint64) *CachedStore {
-	bc.RLock()
-	defer bc.RUnlock()
-	return bc.Stores.TakeStore(containerID)
-}
-
 // PreCheckPutShard checks if the resource is valid to put.
 func (bc *BasicCluster) PreCheckPutShard(res *CachedShard) (*CachedShard, error) {
 	bc.RLock()
@@ -449,13 +442,13 @@ func (bc *BasicCluster) PreCheckPutShard(res *CachedShard) (*CachedShard, error)
 	return origin, nil
 }
 
-// PutShard put a resource, returns overlap resources
+// PutShard put a resource, returns overlap shards
 func (bc *BasicCluster) PutShard(res *CachedShard) []*CachedShard {
 	bc.Lock()
 	defer bc.Unlock()
 
-	if _, ok := bc.WaittingCreateShards[res.Meta.GetID()]; ok {
-		delete(bc.WaittingCreateShards, res.Meta.GetID())
+	if _, ok := bc.WaitingCreateShards[res.Meta.GetID()]; ok {
+		delete(bc.WaitingCreateShards, res.Meta.GetID())
 		if res.Meta.GetState() == metapb.ShardState_Creating {
 			res.Meta.SetState(metapb.ShardState_Running)
 		}
@@ -470,7 +463,7 @@ func (bc *BasicCluster) CheckAndPutShard(res *CachedShard) []*CachedShard {
 		bc.AddRemovedShards(res.Meta.GetID())
 		return nil
 	case metapb.ShardState_Creating:
-		bc.AddWaittingCreateShards(res.Meta)
+		bc.AddWaitingCreateShards(res.Meta)
 		return nil
 	}
 
@@ -484,43 +477,36 @@ func (bc *BasicCluster) CheckAndPutShard(res *CachedShard) []*CachedShard {
 	return bc.PutShard(res)
 }
 
-// RemoveShard removes CachedShard from resourceTree and resourceMap.
+// RemoveShard removes CachedShard from shardTree and shardMap.
 func (bc *BasicCluster) RemoveShard(res *CachedShard) {
 	bc.Lock()
 	defer bc.Unlock()
 	bc.Shards.RemoveShard(res)
 }
 
-// SearchShard searches CachedShard from resourceTree.
+// SearchShard searches CachedShard from shardTree.
 func (bc *BasicCluster) SearchShard(group uint64, resKey []byte) *CachedShard {
 	bc.RLock()
 	defer bc.RUnlock()
 	return bc.Shards.SearchShard(group, resKey)
 }
 
-// SearchPrevShard searches previous CachedShard from resourceTree.
-func (bc *BasicCluster) SearchPrevShard(group uint64, resKey []byte) *CachedShard {
-	bc.RLock()
-	defer bc.RUnlock()
-	return bc.Shards.SearchPrevShard(group, resKey)
-}
-
-// ScanRange scans resources intersecting [start key, end key), returns at most
-// `limit` resources. limit <= 0 means no limit.
+// ScanRange scans shards intersecting [start key, end key), returns at most
+// `limit` shards. limit <= 0 means no limit.
 func (bc *BasicCluster) ScanRange(group uint64, startKey, endKey []byte, limit int) []*CachedShard {
 	bc.RLock()
 	defer bc.RUnlock()
 	return bc.Shards.ScanRange(group, startKey, endKey, limit)
 }
 
-// GetDestroyingShards returns all resources in destroying state
+// GetDestroyingShards returns all shards in destroying state
 func (bc *BasicCluster) GetDestroyingShards() []*CachedShard {
 	bc.RLock()
 	defer bc.RUnlock()
 	return bc.Shards.GetDestroyingShards()
 }
 
-// GetOverlaps returns the resources which are overlapped with the specified resource range.
+// GetOverlaps returns the shards which are overlapped with the specified resource range.
 func (bc *BasicCluster) GetOverlaps(res *CachedShard) []*CachedShard {
 	bc.RLock()
 	defer bc.RUnlock()
@@ -579,7 +565,7 @@ func (bc *BasicCluster) GetShardGroupRuleCount() int {
 	return len(bc.ScheduleGroupRules)
 }
 
-// ShardSetInformer provides access to a shared informer of resources.
+// ShardSetInformer provides access to a shared informer of shards.
 type ShardSetInformer interface {
 	GetScheduleGroupKeys() []string
 	GetScheduleGroupKeysWithPrefix(prefix string) []string
@@ -596,7 +582,7 @@ type ShardSetInformer interface {
 	GetShardByKey(group uint64, resKey []byte) *CachedShard
 }
 
-// StoreSetInformer provides access to a shared informer of containers.
+// StoreSetInformer provides access to a shared informer of stores.
 type StoreSetInformer interface {
 	GetStores() []*CachedStore
 	GetStore(id uint64) *CachedStore
@@ -606,7 +592,7 @@ type StoreSetInformer interface {
 	GetLeaderStore(res *CachedShard) *CachedStore
 }
 
-// StoreSetController is used to control containers' status.
+// StoreSetController is used to control stores' status.
 type StoreSetController interface {
 	PauseLeaderTransfer(id uint64) error
 	ResumeLeaderTransfer(id uint64)
diff --git a/components/prophet/core/kind.go b/components/prophet/core/kind.go
index 6f90b69..f8838ed 100644
--- a/components/prophet/core/kind.go
+++ b/components/prophet/core/kind.go
@@ -28,7 +28,7 @@ const (
 	HighPriority
 )
 
-// ScheduleKind distinguishes resources and schedule policy.
+// ScheduleKind distinguishes shards and schedule policy.
 type ScheduleKind struct {
 	ShardKind metapb.ShardType
 	Policy    SchedulePolicy
@@ -99,17 +99,3 @@ func (k KeyType) String() string {
 		return "unknown"
 	}
 }
-
-// StringToKeyType creates a key type with string.
-func StringToKeyType(input string) KeyType {
-	switch input {
-	case Table.String():
-		return Table
-	case Raw.String():
-		return Raw
-	case Txn.String():
-		return Txn
-	default:
-		panic("invalid key type: " + input)
-	}
-}
diff --git a/components/prophet/core/resource.go b/components/prophet/core/resource.go
deleted file mode 100755
index 4f2ac23..0000000
--- a/components/prophet/core/resource.go
+++ /dev/null
@@ -1,1244 +0,0 @@
-// Copyright 2020 PingCAP, Inc.
-// Modifications copyright (C) 2021 MatrixOrigin.
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//     http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-package core
-
-import (
-	"bytes"
-	"encoding/hex"
-	"fmt"
-	"reflect"
-	"sort"
-	"strings"
-	"sync"
-	"unsafe"
-
-	"github.com/gogo/protobuf/proto"
-	"github.com/matrixorigin/matrixcube/components/prophet/metadata"
-	"github.com/matrixorigin/matrixcube/pb/metapb"
-	"github.com/matrixorigin/matrixcube/pb/rpcpb"
-)
-
-// errShardIsStale is error info for resource is stale.
-var errShardIsStale = func(res metapb.Shard, origin metapb.Shard) error {
-	return fmt.Errorf("resource is stale: resource %v, origin %v", res, origin)
-}
-
-// CachedShard resource runtime info cached in the cache
-type CachedShard struct {
-	sync.RWMutex
-	Meta metapb.Shard
-
-	term            uint64
-	groupKey        string
-	learners        []metapb.Replica
-	voters          []metapb.Replica
-	leader          *metapb.Replica
-	downReplicas    []metapb.ReplicaStats
-	pendingReplicas []metapb.Replica
-	stats           metapb.ShardStats
-}
-
-// NewCachedShard creates CachedShard with resource's meta and leader peer.
-func NewCachedShard(res metapb.Shard, leader *metapb.Replica, opts ...ShardCreateOption) *CachedShard {
-	cr := &CachedShard{
-		Meta:   res,
-		leader: leader,
-	}
-
-	for _, opt := range opts {
-		opt(cr)
-	}
-	classifyVoterAndLearner(cr)
-	return cr
-}
-
-// classifyVoterAndLearner sorts out voter and learner from peers into different slice.
-func classifyVoterAndLearner(res *CachedShard) {
-	learners := make([]metapb.Replica, 0, 1)
-	voters := make([]metapb.Replica, 0, len(res.Meta.GetReplicas()))
-	for _, p := range res.Meta.GetReplicas() {
-		if metadata.IsLearner(p) {
-			learners = append(learners, p)
-		} else {
-			voters = append(voters, p)
-		}
-	}
-	res.learners = learners
-	res.voters = voters
-}
-
-const (
-	// EmptyShardApproximateSize is the resource approximate size of an empty resource
-	// (heartbeat size <= 1MB).
-	EmptyShardApproximateSize = 1
-
-	// ImpossibleFlowSize is an impossible flow size (such as written_bytes, read_keys, etc.)
-	// It may be caused by overflow, refer to https://github.com/tikv/pd/issues/3379.
-	// They need to be filtered so as not to affect downstream.
-	// (flow size >= 1024TB)
-	ImpossibleFlowSize = 1 << 50
-)
-
-// ShardFromHeartbeat constructs a Shard from resource heartbeat.
-func ShardFromHeartbeat(heartbeat rpcpb.ShardHeartbeatReq, meta metapb.Shard) *CachedShard {
-	// Convert unit to MB.
-	// If resource is empty or less than 1MB, use 1MB instead.
-	resourceSize := heartbeat.Stats.GetApproximateSize() / (1 << 20)
-	if resourceSize < EmptyShardApproximateSize {
-		resourceSize = EmptyShardApproximateSize
-	}
-
-	res := &CachedShard{
-		Meta:            meta,
-		groupKey:        heartbeat.GroupKey,
-		term:            heartbeat.GetTerm(),
-		leader:          heartbeat.GetLeader(),
-		downReplicas:    heartbeat.GetDownReplicas(),
-		pendingReplicas: heartbeat.GetPendingReplicas(),
-		stats:           heartbeat.Stats,
-	}
-	res.stats.ApproximateSize = resourceSize
-
-	if res.stats.WrittenKeys >= ImpossibleFlowSize || res.stats.WrittenBytes >= ImpossibleFlowSize {
-		res.stats.WrittenKeys = 0
-		res.stats.WrittenBytes = 0
-	}
-	if res.stats.ReadKeys >= ImpossibleFlowSize || res.stats.ReadBytes >= ImpossibleFlowSize {
-		res.stats.ReadKeys = 0
-		res.stats.ReadBytes = 0
-	}
-
-	sort.Sort(peerStatsSlice(res.downReplicas))
-	sort.Sort(peerSlice(res.pendingReplicas))
-
-	classifyVoterAndLearner(res)
-	return res
-}
-
-// Clone returns a copy of current CachedShard.
-func (r *CachedShard) Clone(opts ...ShardCreateOption) *CachedShard {
-	downReplicas := make([]metapb.ReplicaStats, 0, len(r.downReplicas))
-	for _, peer := range r.downReplicas {
-		downReplicas = append(downReplicas, *(proto.Clone(&peer).(*metapb.ReplicaStats)))
-	}
-	pendingReplicas := make([]metapb.Replica, 0, len(r.pendingReplicas))
-	for _, peer := range r.pendingReplicas {
-		pendingReplicas = append(pendingReplicas, *(proto.Clone(&peer).(*metapb.Replica)))
-	}
-
-	res := &CachedShard{
-		term:            r.term,
-		Meta:            r.Meta,
-		leader:          proto.Clone(r.leader).(*metapb.Replica),
-		downReplicas:    downReplicas,
-		pendingReplicas: pendingReplicas,
-		stats:           r.stats,
-	}
-	res.stats.Interval = proto.Clone(r.stats.Interval).(*metapb.TimeInterval)
-
-	for _, opt := range opts {
-		opt(res)
-	}
-	classifyVoterAndLearner(res)
-	return res
-}
-
-// GetGroupKey returns group key
-func (r *CachedShard) GetGroupKey() string {
-	return r.groupKey
-}
-
-// IsDestroyState resource in Destroyed or Destroying state
-func (r *CachedShard) IsDestroyState() bool {
-	r.RLock()
-	defer r.RUnlock()
-	return r.Meta.GetState() == metapb.ShardState_Destroyed ||
-		r.Meta.GetState() == metapb.ShardState_Destroying
-}
-
-// GetTerm returns the current term of the resource
-func (r *CachedShard) GetTerm() uint64 {
-	return r.term
-}
-
-// GetLearners returns the learners.
-func (r *CachedShard) GetLearners() []metapb.Replica {
-	return r.learners
-}
-
-// GetVoters returns the voters.
-func (r *CachedShard) GetVoters() []metapb.Replica {
-	return r.voters
-}
-
-// GetPeer returns the peer with specified peer id.
-func (r *CachedShard) GetPeer(peerID uint64) (metapb.Replica, bool) {
-	for _, peer := range r.Meta.GetReplicas() {
-		if peer.ID == peerID {
-			return peer, true
-		}
-	}
-	return metapb.Replica{}, false
-}
-
-// GetDownPeer returns the down peer with specified peer id.
-func (r *CachedShard) GetDownPeer(peerID uint64) (metapb.Replica, bool) {
-	for _, down := range r.downReplicas {
-		if down.Replica.ID == peerID {
-			return down.Replica, true
-		}
-	}
-	return metapb.Replica{}, false
-}
-
-// GetDownVoter returns the down voter with specified peer id.
-func (r *CachedShard) GetDownVoter(peerID uint64) (metapb.Replica, bool) {
-	for _, down := range r.downReplicas {
-		if down.Replica.ID == peerID && !metadata.IsLearner(down.Replica) {
-			return down.Replica, true
-		}
-	}
-	return metapb.Replica{}, false
-}
-
-// GetDownLearner returns the down learner with soecified peer id.
-func (r *CachedShard) GetDownLearner(peerID uint64) (metapb.Replica, bool) {
-	for _, down := range r.downReplicas {
-		if down.Replica.ID == peerID && metadata.IsLearner(down.Replica) {
-			return down.Replica, true
-		}
-	}
-	return metapb.Replica{}, false
-}
-
-// GetPendingPeer returns the pending peer with specified peer id.
-func (r *CachedShard) GetPendingPeer(peerID uint64) (metapb.Replica, bool) {
-	for _, peer := range r.pendingReplicas {
-		if peer.ID == peerID {
-			return peer, true
-		}
-	}
-	return metapb.Replica{}, false
-}
-
-// GetPendingVoter returns the pending voter with specified peer id.
-func (r *CachedShard) GetPendingVoter(peerID uint64) (metapb.Replica, bool) {
-	for _, peer := range r.pendingReplicas {
-		if peer.ID == peerID && !metadata.IsLearner(peer) {
-			return peer, true
-		}
-	}
-	return metapb.Replica{}, false
-}
-
-// GetPendingLearner returns the pending learner peer with specified peer id.
-func (r *CachedShard) GetPendingLearner(peerID uint64) (metapb.Replica, bool) {
-	for _, peer := range r.pendingReplicas {
-		if peer.ID == peerID && metadata.IsLearner(peer) {
-			return peer, true
-		}
-	}
-	return metapb.Replica{}, false
-}
-
-// GetStorePeer returns the peer in specified container.
-func (r *CachedShard) GetStorePeer(containerID uint64) (metapb.Replica, bool) {
-	for _, peer := range r.Meta.GetReplicas() {
-		if peer.StoreID == containerID {
-			return peer, true
-		}
-	}
-	return metapb.Replica{}, false
-}
-
-// GetStoreVoter returns the voter in specified container.
-func (r *CachedShard) GetStoreVoter(containerID uint64) (metapb.Replica, bool) {
-	for _, peer := range r.voters {
-		if peer.StoreID == containerID {
-			return peer, true
-		}
-	}
-	return metapb.Replica{}, false
-}
-
-// GetStoreLearner returns the learner peer in specified container.
-func (r *CachedShard) GetStoreLearner(containerID uint64) (metapb.Replica, bool) {
-	for _, peer := range r.learners {
-		if peer.StoreID == containerID {
-			return peer, true
-		}
-	}
-	return metapb.Replica{}, false
-}
-
-// GetStoreIDs returns a map indicate the resource distributed.
-func (r *CachedShard) GetStoreIDs() map[uint64]struct{} {
-	peers := r.Meta.GetReplicas()
-	containerIDs := make(map[uint64]struct{}, len(peers))
-	for _, peer := range peers {
-		containerIDs[peer.StoreID] = struct{}{}
-	}
-	return containerIDs
-}
-
-// GetFollowers returns a map indicate the follow peers distributed.
-func (r *CachedShard) GetFollowers() map[uint64]metapb.Replica {
-	peers := r.GetVoters()
-	followers := make(map[uint64]metapb.Replica, len(peers))
-	for _, peer := range peers {
-		if r.getLeaderID() != peer.ID {
-			followers[peer.StoreID] = peer
-		}
-	}
-	return followers
-}
-
-// GetFollower randomly returns a follow peer.
-func (r *CachedShard) GetFollower() (metapb.Replica, bool) {
-	for _, peer := range r.GetVoters() {
-		if r.getLeaderID() != peer.ID {
-			return peer, true
-		}
-	}
-	return metapb.Replica{}, false
-}
-
-// GetDiffFollowers returns the followers which is not located in the same
-// container as any other followers of the another specified resource.
-func (r *CachedShard) GetDiffFollowers(other *CachedShard) []metapb.Replica {
-	res := make([]metapb.Replica, 0, len(r.Meta.GetReplicas()))
-	for _, p := range r.GetFollowers() {
-		diff := true
-		for _, o := range other.GetFollowers() {
-			if p.StoreID == o.StoreID {
-				diff = false
-				break
-			}
-		}
-		if diff {
-			res = append(res, p)
-		}
-	}
-	return res
-}
-
-// GetStat returns the statistics of the resource.
-func (r *CachedShard) GetStat() *metapb.ShardStats {
-	if r == nil {
-		return nil
-	}
-	stats := r.stats
-	return &stats
-}
-
-// GetApproximateSize returns the approximate size of the resource.
-func (r *CachedShard) GetApproximateSize() int64 {
-	return int64(r.stats.ApproximateSize)
-}
-
-// GetApproximateKeys returns the approximate keys of the resource.
-func (r *CachedShard) GetApproximateKeys() int64 {
-	return int64(r.stats.ApproximateKeys)
-}
-
-// GetInterval returns the interval information of the resource.
-func (r *CachedShard) GetInterval() *metapb.TimeInterval {
-	return r.stats.Interval
-}
-
-// GetDownPeers returns the down peers of the resource.
-func (r *CachedShard) GetDownPeers() []metapb.ReplicaStats {
-	return r.downReplicas
-}
-
-// GetPendingPeers returns the pending peers of the resource.
-func (r *CachedShard) GetPendingPeers() []metapb.Replica {
-	return r.pendingReplicas
-}
-
-// GetBytesRead returns the read bytes of the resource.
-func (r *CachedShard) GetBytesRead() uint64 {
-	return r.stats.ReadBytes
-}
-
-// GetBytesWritten returns the written bytes of the resource.
-func (r *CachedShard) GetBytesWritten() uint64 {
-	return r.stats.WrittenBytes
-}
-
-// GetKeysWritten returns the written keys of the resource.
-func (r *CachedShard) GetKeysWritten() uint64 {
-	return r.stats.WrittenKeys
-}
-
-// GetKeysRead returns the read keys of the resource.
-func (r *CachedShard) GetKeysRead() uint64 {
-	return r.stats.ReadKeys
-}
-
-// GetLeader returns the leader of the resource.
-func (r *CachedShard) GetLeader() *metapb.Replica {
-	return r.leader
-}
-
-// GetStartKey returns the start key of the resource.
-func (r *CachedShard) GetStartKey() []byte {
-	v, _ := r.Meta.GetRange()
-	return v
-}
-
-// GetEndKey returns the end key of the resource.
-func (r *CachedShard) GetEndKey() []byte {
-	_, v := r.Meta.GetRange()
-	return v
-}
-
-func (r *CachedShard) getLeaderID() uint64 {
-	if r.leader == nil {
-		return 0
-	}
-
-	return r.leader.ID
-}
-
-// resourceMap wraps a map[uint64]*CachedShard and supports randomly pick a resource.
-type resourceMap struct {
-	m         map[uint64]*CachedShard
-	totalSize int64
-	totalKeys int64
-}
-
-func newShardMap() *resourceMap {
-	return &resourceMap{
-		m: make(map[uint64]*CachedShard),
-	}
-}
-
-func (rm *resourceMap) Len() int {
-	if rm == nil {
-		return 0
-	}
-	return len(rm.m)
-}
-
-func (rm *resourceMap) Get(id uint64) *CachedShard {
-	if rm == nil {
-		return nil
-	}
-	if r, ok := rm.m[id]; ok {
-		return r
-	}
-	return nil
-}
-
-func (rm *resourceMap) Put(res *CachedShard) {
-	if old, ok := rm.m[res.Meta.GetID()]; ok {
-		rm.totalSize -= int64(old.stats.ApproximateSize)
-		rm.totalKeys -= int64(old.stats.ApproximateKeys)
-	}
-	rm.m[res.Meta.GetID()] = res
-	rm.totalSize += int64(res.stats.ApproximateSize)
-	rm.totalKeys += int64(res.stats.ApproximateKeys)
-}
-
-func (rm *resourceMap) Delete(id uint64) {
-	if rm == nil {
-		return
-	}
-	if old, ok := rm.m[id]; ok {
-		delete(rm.m, id)
-		rm.totalSize -= int64(old.stats.ApproximateSize)
-		rm.totalKeys -= int64(old.stats.ApproximateKeys)
-	}
-}
-
-func (rm *resourceMap) TotalSize() int64 {
-	if rm.Len() == 0 {
-		return 0
-	}
-	return rm.totalSize
-}
-
-// resourceSubTree is used to manager different types of resources.
-type resourceSubTree struct {
-	*resourceTree
-	totalSize int64
-	totalKeys int64
-}
-
-func newShardSubTree() *resourceSubTree {
-	return &resourceSubTree{
-		resourceTree: newShardTree(),
-		totalSize:    0,
-	}
-}
-
-func (rst *resourceSubTree) TotalSize() int64 {
-	if rst.length() == 0 {
-		return 0
-	}
-	return rst.totalSize
-}
-
-func (rst *resourceSubTree) scanRanges() []*CachedShard {
-	if rst.length() == 0 {
-		return nil
-	}
-	var resources []*CachedShard
-	rst.scanRange([]byte(""), func(resource *CachedShard) bool {
-		resources = append(resources, resource)
-		return true
-	})
-	return resources
-}
-
-func (rst *resourceSubTree) update(res *CachedShard) {
-	overlaps := rst.resourceTree.update(res)
-	rst.totalSize += int64(res.stats.ApproximateSize)
-	rst.totalKeys += int64(res.stats.ApproximateKeys)
-	for _, r := range overlaps {
-		rst.totalSize -= int64(r.stats.ApproximateSize)
-		rst.totalKeys -= int64(r.stats.ApproximateKeys)
-	}
-}
-
-func (rst *resourceSubTree) remove(res *CachedShard) {
-	if rst.length() == 0 {
-		return
-	}
-	if rst.resourceTree.remove(res) != nil {
-		rst.totalSize -= int64(res.stats.ApproximateSize)
-		rst.totalKeys -= int64(res.stats.ApproximateKeys)
-	}
-}
-
-func (rst *resourceSubTree) length() int {
-	if rst == nil {
-		return 0
-	}
-	return rst.resourceTree.length()
-}
-
-func (rst *resourceSubTree) RandomShard(ranges []KeyRange) *CachedShard {
-	if rst.length() == 0 {
-		return nil
-	}
-
-	return rst.resourceTree.RandomShard(ranges)
-}
-
-func (rst *resourceSubTree) RandomShards(n int, ranges []KeyRange) []*CachedShard {
-	if rst.length() == 0 {
-		return nil
-	}
-
-	resources := make([]*CachedShard, 0, n)
-
-	for i := 0; i < n; i++ {
-		if resource := rst.resourceTree.RandomShard(ranges); resource != nil {
-			resources = append(resources, resource)
-		}
-	}
-	return resources
-}
-
-// CachedShards for export
-type CachedShards struct {
-	sync.RWMutex
-
-	trees           map[uint64]*resourceTree               // group id -> resourceTree
-	resources       *resourceMap                           // resourceID -> CachedShard
-	leaders         map[string]map[uint64]*resourceSubTree // groupKey -> containerID -> resourceSubTree
-	followers       map[string]map[uint64]*resourceSubTree // groupKey -> containerID -> resourceSubTree
-	learners        map[string]map[uint64]*resourceSubTree // groupKey -> containerID -> resourceSubTree
-	pendingReplicas map[string]map[uint64]*resourceSubTree // groupKey -> containerID -> resourceSubTree
-}
-
-// NewCachedShards creates CachedShards with tree, resources, leaders and followers
-func NewCachedShards() *CachedShards {
-	return &CachedShards{
-		trees:           make(map[uint64]*resourceTree),
-		resources:       newShardMap(),
-		leaders:         make(map[string]map[uint64]*resourceSubTree),
-		followers:       make(map[string]map[uint64]*resourceSubTree),
-		learners:        make(map[string]map[uint64]*resourceSubTree),
-		pendingReplicas: make(map[string]map[uint64]*resourceSubTree),
-	}
-}
-
-func (r *CachedShards) maybeInitWithGroup(groupKey string) {
-	r.Lock()
-	defer r.Unlock()
-
-	if _, ok := r.leaders[groupKey]; !ok {
-		r.leaders[groupKey] = make(map[uint64]*resourceSubTree)
-	}
-	if _, ok := r.followers[groupKey]; !ok {
-		r.followers[groupKey] = make(map[uint64]*resourceSubTree)
-	}
-	if _, ok := r.learners[groupKey]; !ok {
-		r.learners[groupKey] = make(map[uint64]*resourceSubTree)
-	}
-	if _, ok := r.pendingReplicas[groupKey]; !ok {
-		r.pendingReplicas[groupKey] = make(map[uint64]*resourceSubTree)
-	}
-}
-
-// ForeachShards foreach resource by group
-func (r *CachedShards) ForeachShards(group uint64, fn func(res metapb.Shard)) {
-	for _, res := range r.resources.m {
-		if res.Meta.GetGroup() == group {
-			fn(res.Meta)
-		}
-	}
-}
-
-// ForeachCachedShards foreach cached resource by group
-func (r *CachedShards) ForeachCachedShards(group uint64, fn func(res *CachedShard)) {
-	for _, res := range r.resources.m {
-		if res.Meta.GetGroup() == group {
-			fn(res)
-		}
-	}
-}
-
-// GetShard returns the CachedShard with resourceID
-func (r *CachedShards) GetShard(resourceID uint64) *CachedShard {
-	return r.resources.Get(resourceID)
-}
-
-// SetShard sets the CachedShard with resourceID
-func (r *CachedShards) SetShard(res *CachedShard) []*CachedShard {
-	if origin := r.resources.Get(res.Meta.GetID()); origin != nil {
-		if !bytes.Equal(origin.GetStartKey(), res.GetStartKey()) ||
-			!bytes.Equal(origin.GetEndKey(), res.GetEndKey()) {
-			r.removeShardFromTreeAndMap(origin)
-		}
-		if r.shouldRemoveFromSubTree(res, origin) {
-			r.removeShardFromSubTree(origin)
-		}
-	}
-	return r.AddShard(res)
-}
-
-// Length returns the resourcesInfo length
-func (r *CachedShards) Length() int {
-	return r.resources.Len()
-}
-
-// GetOverlaps returns the resources which are overlapped with the specified resource range.
-func (r *CachedShards) GetOverlaps(res *CachedShard) []*CachedShard {
-	if tree, ok := r.trees[res.Meta.GetGroup()]; ok {
-		return tree.getOverlaps(res)
-	}
-
-	return nil
-}
-
-// AddShard adds CachedShard to resourceTree and resourceMap, also update leaders and followers by resource peers
-func (r *CachedShards) AddShard(res *CachedShard) []*CachedShard {
-	if _, ok := r.trees[res.Meta.GetGroup()]; !ok {
-		r.trees[res.Meta.GetGroup()] = newShardTree()
-	}
-
-	// Destroying resource cannot add to tree to avoid range overlaps,
-	// and only wait schedule remove down replicas.
-	if res.IsDestroyState() {
-		r.RemoveShard(res)
-	}
-
-	tree := r.trees[res.Meta.GetGroup()]
-	// the resources which are overlapped with the specified resource range.
-	var overlaps []*CachedShard
-	// when the value is true, add the resource to the tree.
-	// Otherwise use the resource replace the origin resource in the tree.
-	treeNeedAdd := !res.IsDestroyState()
-	if origin := r.GetShard(res.Meta.GetID()); origin != nil {
-		if resOld := tree.find(res); resOld != nil {
-			// Update to tree.
-			if bytes.Equal(resOld.res.GetStartKey(), res.GetStartKey()) &&
-				bytes.Equal(resOld.res.GetEndKey(), res.GetEndKey()) &&
-				resOld.res.Meta.GetID() == res.Meta.GetID() {
-				resOld.res = res
-				treeNeedAdd = false
-			}
-		}
-	}
-	if treeNeedAdd {
-		// Add to tree.
-		overlaps = tree.update(res)
-		for _, item := range overlaps {
-			r.RemoveShard(r.GetShard(item.Meta.GetID()))
-		}
-	}
-	// Add to resources.
-	r.resources.Put(res)
-
-	r.maybeInitWithGroup(res.groupKey)
-
-	// Add to leaders and followers.
-	for _, peer := range res.GetVoters() {
-		containerID := peer.StoreID
-		if peer.ID == res.getLeaderID() {
-			// Add leader peer to leaders.
-			container, ok := r.leaders[res.groupKey][containerID]
-			if !ok {
-				container = newShardSubTree()
-				r.leaders[res.groupKey][containerID] = container
-			}
-			container.update(res)
-		} else {
-			// Add follower peer to followers.
-			container, ok := r.followers[res.groupKey][containerID]
-			if !ok {
-				container = newShardSubTree()
-				r.followers[res.groupKey][containerID] = container
-			}
-			container.update(res)
-		}
-	}
-
-	// Add to learners.
-	for _, peer := range res.GetLearners() {
-		containerID := peer.StoreID
-		container, ok := r.learners[res.groupKey][containerID]
-		if !ok {
-			container = newShardSubTree()
-			r.learners[res.groupKey][containerID] = container
-		}
-		container.update(res)
-	}
-
-	for _, peer := range res.pendingReplicas {
-		containerID := peer.StoreID
-		container, ok := r.pendingReplicas[res.groupKey][containerID]
-		if !ok {
-			container = newShardSubTree()
-			r.pendingReplicas[res.groupKey][containerID] = container
-		}
-		container.update(res)
-	}
-
-	return overlaps
-}
-
-// RemoveShard removes CachedShard from resourceTree and resourceMap
-func (r *CachedShards) RemoveShard(res *CachedShard) {
-	// Remove from tree and resources.
-	r.removeShardFromTreeAndMap(res)
-	// Remove from leaders and followers.
-	r.removeShardFromSubTree(res)
-}
-
-// removeShardFromTreeAndMap removes CachedShard from resourceTree and resourceMap
-func (r *CachedShards) removeShardFromTreeAndMap(res *CachedShard) {
-	// Remove from tree and resources.
-	if tree, ok := r.trees[res.Meta.GetGroup()]; ok {
-		tree.remove(res)
-	}
-	r.resources.Delete(res.Meta.GetID())
-}
-
-// removeShardFromSubTree removes CachedShard from resourcesubTrees
-func (r *CachedShards) removeShardFromSubTree(res *CachedShard) {
-	r.maybeInitWithGroup(res.groupKey)
-
-	// Remove from leaders and followers.
-	for _, peer := range res.Meta.GetReplicas() {
-		containerID := peer.StoreID
-		r.leaders[res.groupKey][containerID].remove(res)
-		r.followers[res.groupKey][containerID].remove(res)
-		r.learners[res.groupKey][containerID].remove(res)
-		r.pendingReplicas[res.groupKey][containerID].remove(res)
-	}
-}
-
-type peerSlice []metapb.Replica
-
-func (s peerSlice) Len() int {
-	return len(s)
-}
-func (s peerSlice) Swap(i, j int) {
-	s[i], s[j] = s[j], s[i]
-}
-func (s peerSlice) Less(i, j int) bool {
-	return s[i].ID < s[j].ID
-}
-
-// SortedPeersEqual judges whether two sorted `peerSlice` are equal
-func SortedPeersEqual(peersA, peersB []metapb.Replica) bool {
-	if len(peersA) != len(peersB) {
-		return false
-	}
-	for i, peer := range peersA {
-		if peer.GetID() != peersB[i].GetID() {
-			return false
-		}
-	}
-	return true
-}
-
-type peerStatsSlice []metapb.ReplicaStats
-
-func (s peerStatsSlice) Len() int {
-	return len(s)
-}
-func (s peerStatsSlice) Swap(i, j int) {
-	s[i], s[j] = s[j], s[i]
-}
-func (s peerStatsSlice) Less(i, j int) bool {
-	return s[i].GetReplica().ID < s[j].GetReplica().ID
-}
-
-// SortedPeersStatsEqual judges whether two sorted `peerStatsSlice` are equal
-func SortedPeersStatsEqual(peersA, peersB []metapb.ReplicaStats) bool {
-	if len(peersA) != len(peersB) {
-		return false
-	}
-	for i, peerStats := range peersA {
-		if peerStats.GetReplica().ID != peersB[i].GetReplica().ID ||
-			peerStats.GetDownSeconds() != peersB[i].GetDownSeconds() {
-			return false
-		}
-	}
-	return true
-}
-
-// shouldRemoveFromSubTree return true when the resource leader changed, peer transferred,
-// new peer was created, learners changed, pendingReplicas changed, and so on.
-func (r *CachedShards) shouldRemoveFromSubTree(res *CachedShard, origin *CachedShard) bool {
-	checkPeersChange := func(origin []metapb.Replica, other []metapb.Replica) bool {
-		if len(origin) != len(other) {
-			return true
-		}
-		sort.Sort(peerSlice(origin))
-		sort.Sort(peerSlice(other))
-		for index, peer := range origin {
-			if peer.StoreID == other[index].StoreID && peer.ID == other[index].ID {
-				continue
-			}
-			return true
-		}
-		return false
-	}
-
-	return origin.getLeaderID() != res.getLeaderID() ||
-		checkPeersChange(origin.GetVoters(), res.GetVoters()) ||
-		checkPeersChange(origin.GetLearners(), res.GetLearners()) ||
-		checkPeersChange(origin.GetPendingPeers(), res.GetPendingPeers()) ||
-		origin.groupKey != res.groupKey
-}
-
-// SearchShard searches CachedShard from resourceTree
-func (r *CachedShards) SearchShard(group uint64, resKey []byte) *CachedShard {
-	if tree, ok := r.trees[group]; ok {
-		res := tree.search(resKey)
-		if res == nil {
-			return nil
-		}
-		return r.GetShard(res.Meta.GetID())
-	}
-
-	return nil
-}
-
-// SearchPrevShard searches previous CachedShard from resourceTree
-func (r *CachedShards) SearchPrevShard(group uint64, resKey []byte) *CachedShard {
-	if tree, ok := r.trees[group]; ok {
-		res := tree.searchPrev(resKey)
-		if res == nil {
-			return nil
-		}
-		return r.GetShard(res.Meta.GetID())
-	}
-
-	return nil
-}
-
-// GetShards gets all CachedShard from resourceMap
-func (r *CachedShards) GetShards() []*CachedShard {
-	resources := make([]*CachedShard, 0, r.resources.Len())
-	for _, res := range r.resources.m {
-		resources = append(resources, res)
-	}
-	return resources
-}
-
-// GetStoreShards gets all CachedShard with a given containerID
-func (r *CachedShards) GetStoreShards(groupKey string, containerID uint64) []*CachedShard {
-	r.maybeInitWithGroup(groupKey)
-	resources := make([]*CachedShard, 0, r.GetStoreShardCount(groupKey, containerID))
-	if leaders, ok := r.leaders[groupKey][containerID]; ok {
-		resources = append(resources, leaders.scanRanges()...)
-	}
-	if followers, ok := r.followers[groupKey][containerID]; ok {
-		resources = append(resources, followers.scanRanges()...)
-	}
-	if learners, ok := r.learners[groupKey][containerID]; ok {
-		resources = append(resources, learners.scanRanges()...)
-	}
-	return resources
-}
-
-// GetStoreLeaderShardSize get total size of container's leader resources
-func (r *CachedShards) GetStoreLeaderShardSize(groupKey string, containerID uint64) int64 {
-	r.maybeInitWithGroup(groupKey)
-	return r.leaders[groupKey][containerID].TotalSize()
-}
-
-// GetStoreFollowerShardSize get total size of container's follower resources
-func (r *CachedShards) GetStoreFollowerShardSize(groupKey string, containerID uint64) int64 {
-	r.maybeInitWithGroup(groupKey)
-	return r.followers[groupKey][containerID].TotalSize()
-}
-
-// GetStoreLearnerShardSize get total size of container's learner resources
-func (r *CachedShards) GetStoreLearnerShardSize(groupKey string, containerID uint64) int64 {
-	r.maybeInitWithGroup(groupKey)
-	return r.learners[groupKey][containerID].TotalSize()
-}
-
-// GetStoreShardSize get total size of container's resources
-func (r *CachedShards) GetStoreShardSize(groupKey string, containerID uint64) int64 {
-	r.maybeInitWithGroup(groupKey)
-	return r.GetStoreLeaderShardSize(groupKey, containerID) +
-		r.GetStoreFollowerShardSize(groupKey, containerID) +
-		r.GetStoreLearnerShardSize(groupKey, containerID)
-}
-
-// GetMetaShards gets a set of *metapb.Shard from resourceMap
-func (r *CachedShards) GetMetaShards() []*metapb.Shard {
-	resources := make([]*metapb.Shard, 0, r.resources.Len())
-	for _, res := range r.resources.m {
-		resources = append(resources, res.Meta.Clone())
-	}
-	return resources
-}
-
-// GetShardCount gets the total count of CachedShard of resourceMap
-func (r *CachedShards) GetShardCount() int {
-	return r.resources.Len()
-}
-
-// GetStoreShardCount gets the total count of a container's leader, follower and learner CachedShard by containerID
-func (r *CachedShards) GetStoreShardCount(groupKey string, containerID uint64) int {
-	r.maybeInitWithGroup(groupKey)
-	return r.GetStoreLeaderCount(groupKey, containerID) +
-		r.GetStoreFollowerCount(groupKey, containerID) +
-		r.GetStoreLearnerCount(groupKey, containerID)
-}
-
-// GetStorePendingPeerCount gets the total count of a container's resource that includes pending peer
-func (r *CachedShards) GetStorePendingPeerCount(groupKey string, containerID uint64) int {
-	r.maybeInitWithGroup(groupKey)
-	return r.pendingReplicas[groupKey][containerID].length()
-}
-
-// GetStoreLeaderCount get the total count of a container's leader CachedShard
-func (r *CachedShards) GetStoreLeaderCount(groupKey string, containerID uint64) int {
-	r.maybeInitWithGroup(groupKey)
-	return r.leaders[groupKey][containerID].length()
-}
-
-// GetStoreFollowerCount get the total count of a container's follower CachedShard
-func (r *CachedShards) GetStoreFollowerCount(groupKey string, containerID uint64) int {
-	r.maybeInitWithGroup(groupKey)
-	return r.followers[groupKey][containerID].length()
-}
-
-// GetStoreLearnerCount get the total count of a container's learner CachedShard
-func (r *CachedShards) GetStoreLearnerCount(groupKey string, containerID uint64) int {
-	r.maybeInitWithGroup(groupKey)
-	return r.learners[groupKey][containerID].length()
-}
-
-// RandPendingShard randomly gets a container's resource with a pending peer.
-func (r *CachedShards) RandPendingShard(groupKey string, containerID uint64, ranges []KeyRange) *CachedShard {
-	r.maybeInitWithGroup(groupKey)
-	return r.pendingReplicas[groupKey][containerID].RandomShard(ranges)
-}
-
-// RandPendingShards randomly gets a container's n resources with a pending peer.
-func (r *CachedShards) RandPendingShards(groupKey string, containerID uint64, ranges []KeyRange, n int) []*CachedShard {
-	r.maybeInitWithGroup(groupKey)
-	return r.pendingReplicas[groupKey][containerID].RandomShards(n, ranges)
-}
-
-// RandLeaderShard randomly gets a container's leader resource.
-func (r *CachedShards) RandLeaderShard(groupKey string, containerID uint64, ranges []KeyRange) *CachedShard {
-	r.maybeInitWithGroup(groupKey)
-	return r.leaders[groupKey][containerID].RandomShard(ranges)
-}
-
-// RandLeaderShards randomly gets a container's n leader resources.
-func (r *CachedShards) RandLeaderShards(groupKey string, containerID uint64, ranges []KeyRange, n int) []*CachedShard {
-	r.maybeInitWithGroup(groupKey)
-	return r.leaders[groupKey][containerID].RandomShards(n, ranges)
-}
-
-// RandFollowerShard randomly gets a container's follower resource.
-func (r *CachedShards) RandFollowerShard(groupKey string, containerID uint64, ranges []KeyRange) *CachedShard {
-	r.maybeInitWithGroup(groupKey)
-	return r.followers[groupKey][containerID].RandomShard(ranges)
-}
-
-// RandFollowerShards randomly gets a container's n follower resources.
-func (r *CachedShards) RandFollowerShards(groupKey string, containerID uint64, ranges []KeyRange, n int) []*CachedShard {
-	r.maybeInitWithGroup(groupKey)
-	return r.followers[groupKey][containerID].RandomShards(n, ranges)
-}
-
-// RandLearnerShard randomly gets a container's learner resource.
-func (r *CachedShards) RandLearnerShard(groupKey string, containerID uint64, ranges []KeyRange) *CachedShard {
-	r.maybeInitWithGroup(groupKey)
-	return r.learners[groupKey][containerID].RandomShard(ranges)
-}
-
-// RandLearnerShards randomly gets a container's n learner resources.
-func (r *CachedShards) RandLearnerShards(groupKey string, containerID uint64, ranges []KeyRange, n int) []*CachedShard {
-	r.maybeInitWithGroup(groupKey)
-	return r.learners[groupKey][containerID].RandomShards(n, ranges)
-}
-
-// GetLeader return leader CachedShard by containerID and resourceID(now only used in test)
-func (r *CachedShards) GetLeader(groupKey string, containerID uint64, res *CachedShard) *CachedShard {
-	r.maybeInitWithGroup(groupKey)
-	if leaders, ok := r.leaders[groupKey][containerID]; ok {
-		return leaders.find(res).res
-	}
-	return nil
-}
-
-// GetFollower return follower CachedShard by containerID and resourceID(now only used in test)
-func (r *CachedShards) GetFollower(groupKey string, containerID uint64, res *CachedShard) *CachedShard {
-	r.maybeInitWithGroup(groupKey)
-	if followers, ok := r.followers[groupKey][containerID]; ok {
-		return followers.find(res).res
-	}
-	return nil
-}
-
-// ScanRange scans resources intersecting [start key, end key), returns at most
-// `limit` resources. limit <= 0 means no limit.
-func (r *CachedShards) ScanRange(group uint64, startKey, endKey []byte, limit int) []*CachedShard {
-	var resources []*CachedShard
-	if tree, ok := r.trees[group]; ok {
-		tree.scanRange(startKey, func(resource *CachedShard) bool {
-			if len(endKey) > 0 && bytes.Compare(resource.GetStartKey(), endKey) >= 0 {
-				return false
-			}
-			if limit > 0 && len(resources) >= limit {
-				return false
-			}
-			resources = append(resources, r.GetShard(resource.Meta.GetID()))
-			return true
-		})
-	}
-	return resources
-}
-
-// GetDestroyingShards returns all resources in destroying state
-func (r *CachedShards) GetDestroyingShards() []*CachedShard {
-	var resources []*CachedShard
-	for _, res := range r.resources.m {
-		if res.Meta.GetState() == metapb.ShardState_Destroying {
-			resources = append(resources, res)
-		}
-	}
-	return resources
-}
-
-// ScanRangeWithIterator scans from the first resource containing or behind start key,
-// until iterator returns false.
-func (r *CachedShards) ScanRangeWithIterator(group uint64, startKey []byte, iterator func(res *CachedShard) bool) {
-	if tree, ok := r.trees[group]; ok {
-		tree.scanRange(startKey, iterator)
-	}
-}
-
-// GetAdjacentShards returns resource's info that is adjacent with specific resource
-func (r *CachedShards) GetAdjacentShards(res *CachedShard) (*CachedShard, *CachedShard) {
-	var prev, next *CachedShard
-	if tree, ok := r.trees[res.Meta.GetGroup()]; ok {
-		p, n := tree.getAdjacentShards(res)
-		// check key to avoid key range hole
-		if p != nil && bytes.Equal(p.res.GetEndKey(), res.GetStartKey()) {
-			prev = r.GetShard(p.res.Meta.GetID())
-		}
-		if n != nil && bytes.Equal(res.GetEndKey(), n.res.GetStartKey()) {
-			next = r.GetShard(n.res.Meta.GetID())
-		}
-	}
-	return prev, next
-}
-
-// GetAverageShardSize returns the average resource approximate size.
-func (r *CachedShards) GetAverageShardSize() int64 {
-	if r.resources.Len() == 0 {
-		return 0
-	}
-	return r.resources.TotalSize() / int64(r.resources.Len())
-}
-
-// DiffShardPeersInfo return the difference of peers info  between two CachedShard
-func DiffShardPeersInfo(origin *CachedShard, other *CachedShard) string {
-	var ret []string
-	for _, a := range origin.Meta.GetReplicas() {
-		both := false
-		for _, b := range other.Meta.GetReplicas() {
-			if reflect.DeepEqual(a, b) {
-				both = true
-				break
-			}
-		}
-		if !both {
-			ret = append(ret, fmt.Sprintf("Remove peer:{%v}", a))
-		}
-	}
-	for _, b := range other.Meta.GetReplicas() {
-		both := false
-		for _, a := range origin.Meta.GetReplicas() {
-			if reflect.DeepEqual(a, b) {
-				both = true
-				break
-			}
-		}
-		if !both {
-			ret = append(ret, fmt.Sprintf("Add peer:{%v}", b))
-		}
-	}
-	return strings.Join(ret, ",")
-}
-
-// DiffShardKeyInfo return the difference of key info between two CachedShard
-func DiffShardKeyInfo(origin *CachedShard, other *CachedShard) string {
-	originStartKey, originEndKey := origin.Meta.GetRange()
-	otherStartKey, otherEndKey := other.Meta.GetRange()
-
-	var ret []string
-	if !bytes.Equal(originStartKey, otherStartKey) {
-		ret = append(ret, fmt.Sprintf("StartKey Changed:{%s} -> {%s}", HexShardKey(originStartKey), HexShardKey(otherStartKey)))
-	} else {
-		ret = append(ret, fmt.Sprintf("StartKey:{%s}", HexShardKey(originStartKey)))
-	}
-	if !bytes.Equal(originEndKey, otherEndKey) {
-		ret = append(ret, fmt.Sprintf("EndKey Changed:{%s} -> {%s}", HexShardKey(originEndKey), HexShardKey(otherEndKey)))
-	} else {
-		ret = append(ret, fmt.Sprintf("EndKey:{%s}", HexShardKey(originEndKey)))
-	}
-
-	return strings.Join(ret, ", ")
-}
-
-func isInvolved(res *CachedShard, startKey, endKey []byte) bool {
-	return bytes.Compare(res.GetStartKey(), startKey) >= 0 && (len(endKey) == 0 || (len(res.GetEndKey()) > 0 && bytes.Compare(res.GetEndKey(), endKey) <= 0))
-}
-
-// String converts slice of bytes to string without copy.
-func String(b []byte) (s string) {
-	if len(b) == 0 {
-		return ""
-	}
-	pbytes := (*reflect.SliceHeader)(unsafe.Pointer(&b))
-	pstring := (*reflect.StringHeader)(unsafe.Pointer(&s))
-	pstring.Data = pbytes.Data
-	pstring.Len = pbytes.Len
-	return
-}
-
-// ToUpperASCIIInplace bytes.ToUpper but zero-cost
-func ToUpperASCIIInplace(s []byte) []byte {
-	hasLower := false
-	for i := 0; i < len(s); i++ {
-		c := s[i]
-		hasLower = hasLower || ('a' <= c && c <= 'z')
-	}
-
-	if !hasLower {
-		return s
-	}
-	var c byte
-	for i := 0; i < len(s); i++ {
-		c = s[i]
-		if 'a' <= c && c <= 'z' {
-			c -= 'a' - 'A'
-		}
-		s[i] = c
-	}
-	return s
-}
-
-// EncodeToString overrides hex.EncodeToString implementation. Difference: returns []byte, not string
-func EncodeToString(src []byte) []byte {
-	dst := make([]byte, hex.EncodedLen(len(src)))
-	hex.Encode(dst, src)
-	return dst
-}
-
-// HexShardKey converts resource key to hex format. Used for formating resource in
-// logs.
-func HexShardKey(key []byte) []byte {
-	return ToUpperASCIIInplace(EncodeToString(key))
-}
-
-// HexShardKeyStr converts resource key to hex format. Used for formating resource in
-// logs.
-func HexShardKeyStr(key []byte) string {
-	return String(HexShardKey(key))
-}
-
-// ShardToHexMeta converts a resource meta's keys to hex format. Used for formating
-// resource in logs.
-func ShardToHexMeta(meta metapb.Shard) HexShardMeta {
-	start, end := meta.GetRange()
-	meta.SetStartKey(HexShardKey(start))
-	meta.SetEndKey(HexShardKey(end))
-	return HexShardMeta{meta}
-}
-
-// HexShardMeta is a resource meta in the hex format. Used for formating resource in logs.
-type HexShardMeta struct {
-	meta metapb.Shard
-}
-
-func (h HexShardMeta) String() string {
-	return fmt.Sprintf("resource %+v", h.meta)
-}
-
-// ShardsToHexMeta converts resources' meta keys to hex format. Used for formating
-// resource in logs.
-func ShardsToHexMeta(resources []*metapb.Shard) HexShardsMeta {
-	hexShardMetas := make([]*metapb.Shard, len(resources))
-	for i, res := range resources {
-		meta := res.Clone()
-		start, end := meta.GetRange()
-		meta.SetStartKey(HexShardKey(start))
-		meta.SetEndKey(HexShardKey(end))
-		hexShardMetas[i] = meta
-	}
-	return hexShardMetas
-}
-
-// HexShardsMeta is a slice of resources' meta in the hex format. Used for formating
-// resource in logs.
-type HexShardsMeta []*metapb.Shard
-
-func (h HexShardsMeta) String() string {
-	var b strings.Builder
-	for _, r := range h {
-		b.WriteString(fmt.Sprintf("resource %+v", r))
-	}
-	return strings.TrimSpace(b.String())
-}
diff --git a/components/prophet/core/shard.go b/components/prophet/core/shard.go
new file mode 100755
index 0000000..9ed03d1
--- /dev/null
+++ b/components/prophet/core/shard.go
@@ -0,0 +1,1180 @@
+// Copyright 2020 PingCAP, Inc.
+// Modifications copyright (C) 2021 MatrixOrigin.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package core
+
+import (
+	"bytes"
+	"encoding/hex"
+	"fmt"
+	"reflect"
+	"sort"
+	"strings"
+	"sync"
+	"unsafe"
+
+	"github.com/gogo/protobuf/proto"
+	"github.com/matrixorigin/matrixcube/components/prophet/metadata"
+	"github.com/matrixorigin/matrixcube/pb/metapb"
+	"github.com/matrixorigin/matrixcube/pb/rpcpb"
+)
+
+// errShardIsStale is error info for shard is stale.
+var errShardIsStale = func(res metapb.Shard, origin metapb.Shard) error {
+	return fmt.Errorf("shard is stale: shard %v, origin %v", res, origin)
+}
+
+// CachedShard shard runtime info cached in the cache
+type CachedShard struct {
+	sync.RWMutex
+	Meta metapb.Shard
+
+	term            uint64
+	groupKey        string
+	learners        []metapb.Replica
+	voters          []metapb.Replica
+	leader          *metapb.Replica
+	downReplicas    replicaStatsSlice
+	pendingReplicas replicaSlice
+	stats           metapb.ShardStats
+}
+
+// NewCachedShard creates CachedShard with shard's meta and leader peer.
+func NewCachedShard(res metapb.Shard, leader *metapb.Replica, opts ...ShardCreateOption) *CachedShard {
+	cr := &CachedShard{
+		Meta:   res,
+		leader: leader,
+	}
+
+	for _, opt := range opts {
+		opt(cr)
+	}
+	fillVoterAndLearner(cr)
+	return cr
+}
+
+// fillVoterAndLearner sorts out voter and learner from peers into different slice.
+func fillVoterAndLearner(res *CachedShard) {
+	learners := make([]metapb.Replica, 0, 1)
+	voters := make([]metapb.Replica, 0, len(res.Meta.GetReplicas()))
+	for _, p := range res.Meta.GetReplicas() {
+		if metadata.IsLearner(p) {
+			learners = append(learners, p)
+		} else {
+			voters = append(voters, p)
+		}
+	}
+	res.learners = learners
+	res.voters = voters
+}
+
+const (
+	// MinShardSize is the shard approximate size of an empty shard
+	// (heartbeat size <= 1MB).
+	MinShardSize = 1
+
+	// MaximumFlowSize is an impossible flow size (such as written_bytes, read_keys, etc.)
+	// It may be caused by overflow, refer to https://github.com/tikv/pd/issues/3379.
+	// They need to be filtered so as not to affect downstream.
+	// (flow size >= 1024TB)
+	MaximumFlowSize = 1 << 50
+)
+
+// ShardFromHeartbeat constructs a Shard from shard heartbeat.
+func ShardFromHeartbeat(heartbeat rpcpb.ShardHeartbeatReq, meta metapb.Shard) *CachedShard {
+	// Convert unit to MB.
+	// If shard is empty or less than 1MB, use 1MB instead.
+	shardSize := heartbeat.Stats.GetApproximateSize() / (1 << 20)
+	if shardSize < MinShardSize {
+		shardSize = MinShardSize
+	}
+
+	shard := &CachedShard{
+		Meta:            meta,
+		groupKey:        heartbeat.GroupKey,
+		term:            heartbeat.GetTerm(),
+		leader:          heartbeat.GetLeader(),
+		downReplicas:    heartbeat.GetDownReplicas(),
+		pendingReplicas: heartbeat.GetPendingReplicas(),
+		stats:           heartbeat.Stats,
+	}
+	shard.stats.ApproximateSize = shardSize
+
+	if shard.stats.WrittenKeys >= MaximumFlowSize || shard.stats.WrittenBytes >= MaximumFlowSize {
+		shard.stats.WrittenKeys = 0
+		shard.stats.WrittenBytes = 0
+	}
+	if shard.stats.ReadKeys >= MaximumFlowSize || shard.stats.ReadBytes >= MaximumFlowSize {
+		shard.stats.ReadKeys = 0
+		shard.stats.ReadBytes = 0
+	}
+
+	sort.Sort(shard.downReplicas)
+	sort.Sort(shard.pendingReplicas)
+
+	fillVoterAndLearner(shard)
+	return shard
+}
+
+// Clone returns a copy of current CachedShard.
+func (r *CachedShard) Clone(opts ...ShardCreateOption) *CachedShard {
+	downReplicas := make([]metapb.ReplicaStats, 0, len(r.downReplicas))
+	for _, peer := range r.downReplicas {
+		downReplicas = append(downReplicas, *(proto.Clone(&peer).(*metapb.ReplicaStats)))
+	}
+	pendingReplicas := make([]metapb.Replica, 0, len(r.pendingReplicas))
+	for _, peer := range r.pendingReplicas {
+		pendingReplicas = append(pendingReplicas, *(proto.Clone(&peer).(*metapb.Replica)))
+	}
+
+	res := &CachedShard{
+		term:            r.term,
+		Meta:            r.Meta,
+		leader:          proto.Clone(r.leader).(*metapb.Replica),
+		downReplicas:    downReplicas,
+		pendingReplicas: pendingReplicas,
+		stats:           r.stats,
+	}
+	res.stats.Interval = proto.Clone(r.stats.Interval).(*metapb.TimeInterval)
+
+	for _, opt := range opts {
+		opt(res)
+	}
+	fillVoterAndLearner(res)
+	return res
+}
+
+// GetGroupKey returns group key
+func (r *CachedShard) GetGroupKey() string {
+	return r.groupKey
+}
+
+// IsDestroyState shard in Destroyed or Destroying state
+func (r *CachedShard) IsDestroyState() bool {
+	r.RLock()
+	defer r.RUnlock()
+	return r.Meta.GetState() == metapb.ShardState_Destroyed ||
+		r.Meta.GetState() == metapb.ShardState_Destroying
+}
+
+// GetTerm returns the current term of the shard
+func (r *CachedShard) GetTerm() uint64 {
+	return r.term
+}
+
+// GetLearners returns the learners.
+func (r *CachedShard) GetLearners() []metapb.Replica {
+	return r.learners
+}
+
+// GetVoters returns the voters.
+func (r *CachedShard) GetVoters() []metapb.Replica {
+	return r.voters
+}
+
+// GetPeer returns the peer with specified peer id.
+func (r *CachedShard) GetPeer(peerID uint64) (metapb.Replica, bool) {
+	for _, peer := range r.Meta.GetReplicas() {
+		if peer.ID == peerID {
+			return peer, true
+		}
+	}
+	return metapb.Replica{}, false
+}
+
+// GetDownPeer returns the down peer with specified peer id.
+func (r *CachedShard) GetDownPeer(peerID uint64) (metapb.Replica, bool) {
+	for _, down := range r.downReplicas {
+		if down.Replica.ID == peerID {
+			return down.Replica, true
+		}
+	}
+	return metapb.Replica{}, false
+}
+
+// GetDownVoter returns the down voter with specified peer id.
+func (r *CachedShard) GetDownVoter(peerID uint64) (metapb.Replica, bool) {
+	for _, down := range r.downReplicas {
+		if down.Replica.ID == peerID && !metadata.IsLearner(down.Replica) {
+			return down.Replica, true
+		}
+	}
+	return metapb.Replica{}, false
+}
+
+// GetDownLearner returns the down learner with specified peer id.
+func (r *CachedShard) GetDownLearner(peerID uint64) (metapb.Replica, bool) {
+	for _, down := range r.downReplicas {
+		if down.Replica.ID == peerID && metadata.IsLearner(down.Replica) {
+			return down.Replica, true
+		}
+	}
+	return metapb.Replica{}, false
+}
+
+// GetPendingPeer returns the pending peer with specified peer id.
+func (r *CachedShard) GetPendingPeer(peerID uint64) (metapb.Replica, bool) {
+	for _, peer := range r.pendingReplicas {
+		if peer.ID == peerID {
+			return peer, true
+		}
+	}
+	return metapb.Replica{}, false
+}
+
+// GetPendingVoter returns the pending voter with specified peer id.
+func (r *CachedShard) GetPendingVoter(peerID uint64) (metapb.Replica, bool) {
+	for _, peer := range r.pendingReplicas {
+		if peer.ID == peerID && !metadata.IsLearner(peer) {
+			return peer, true
+		}
+	}
+	return metapb.Replica{}, false
+}
+
+// GetPendingLearner returns the pending learner peer with specified peer id.
+func (r *CachedShard) GetPendingLearner(peerID uint64) (metapb.Replica, bool) {
+	for _, peer := range r.pendingReplicas {
+		if peer.ID == peerID && metadata.IsLearner(peer) {
+			return peer, true
+		}
+	}
+	return metapb.Replica{}, false
+}
+
+// GetStorePeer returns the peer in specified container.
+func (r *CachedShard) GetStorePeer(storeID uint64) (metapb.Replica, bool) {
+	for _, peer := range r.Meta.GetReplicas() {
+		if peer.StoreID == storeID {
+			return peer, true
+		}
+	}
+	return metapb.Replica{}, false
+}
+
+// GetStoreVoter returns the voter in specified container.
+func (r *CachedShard) GetStoreVoter(storeID uint64) (metapb.Replica, bool) {
+	for _, peer := range r.voters {
+		if peer.StoreID == storeID {
+			return peer, true
+		}
+	}
+	return metapb.Replica{}, false
+}
+
+// GetStoreLearner returns the learner peer in specified container.
+func (r *CachedShard) GetStoreLearner(storeID uint64) (metapb.Replica, bool) {
+	for _, peer := range r.learners {
+		if peer.StoreID == storeID {
+			return peer, true
+		}
+	}
+	return metapb.Replica{}, false
+}
+
+// GetStoreIDs returns a map indicate the shard distributed.
+func (r *CachedShard) GetStoreIDs() map[uint64]struct{} {
+	peers := r.Meta.GetReplicas()
+	storeIDs := make(map[uint64]struct{}, len(peers))
+	for _, peer := range peers {
+		storeIDs[peer.StoreID] = struct{}{}
+	}
+	return storeIDs
+}
+
+// GetFollowers returns a map indicate the follow peers distributed.
+func (r *CachedShard) GetFollowers() map[uint64]metapb.Replica {
+	peers := r.GetVoters()
+	followers := make(map[uint64]metapb.Replica, len(peers))
+	for _, peer := range peers {
+		if r.getLeaderID() != peer.ID {
+			followers[peer.StoreID] = peer
+		}
+	}
+	return followers
+}
+
+// GetFollower randomly returns a follow peer.
+func (r *CachedShard) GetFollower() (metapb.Replica, bool) {
+	for _, peer := range r.GetVoters() {
+		if r.getLeaderID() != peer.ID {
+			return peer, true
+		}
+	}
+	return metapb.Replica{}, false
+}
+
+// GetDiffFollowers returns the followers which is not located in the same
+// container as any other followers of the another specified shard.
+func (r *CachedShard) GetDiffFollowers(other *CachedShard) []metapb.Replica {
+	res := make([]metapb.Replica, 0, len(r.Meta.GetReplicas()))
+	for _, p := range r.GetFollowers() {
+		diff := true
+		for _, o := range other.GetFollowers() {
+			if p.StoreID == o.StoreID {
+				diff = false
+				break
+			}
+		}
+		if diff {
+			res = append(res, p)
+		}
+	}
+	return res
+}
+
+// GetStat returns the statistics of the shard.
+func (r *CachedShard) GetStat() *metapb.ShardStats {
+	if r == nil {
+		return nil
+	}
+	stats := r.stats
+	return &stats
+}
+
+// GetApproximateSize returns the approximate size of the shard.
+func (r *CachedShard) GetApproximateSize() int64 {
+	return int64(r.stats.ApproximateSize)
+}
+
+// GetApproximateKeys returns the approximate keys of the shard.
+func (r *CachedShard) GetApproximateKeys() int64 {
+	return int64(r.stats.ApproximateKeys)
+}
+
+// GetInterval returns the interval information of the shard.
+func (r *CachedShard) GetInterval() *metapb.TimeInterval {
+	return r.stats.Interval
+}
+
+// GetDownPeers returns the down peers of the shard.
+func (r *CachedShard) GetDownPeers() []metapb.ReplicaStats {
+	return r.downReplicas
+}
+
+// GetPendingPeers returns the pending peers of the shard.
+func (r *CachedShard) GetPendingPeers() []metapb.Replica {
+	return r.pendingReplicas
+}
+
+// GetBytesRead returns the read bytes of the shard.
+func (r *CachedShard) GetBytesRead() uint64 {
+	return r.stats.ReadBytes
+}
+
+// GetBytesWritten returns the written bytes of the shard.
+func (r *CachedShard) GetBytesWritten() uint64 {
+	return r.stats.WrittenBytes
+}
+
+// GetKeysWritten returns the written keys of the shard.
+func (r *CachedShard) GetKeysWritten() uint64 {
+	return r.stats.WrittenKeys
+}
+
+// GetKeysRead returns the read keys of the shard.
+func (r *CachedShard) GetKeysRead() uint64 {
+	return r.stats.ReadKeys
+}
+
+// GetLeader returns the leader of the shard.
+func (r *CachedShard) GetLeader() *metapb.Replica {
+	return r.leader
+}
+
+// GetStartKey returns the start key of the shard.
+func (r *CachedShard) GetStartKey() []byte {
+	v, _ := r.Meta.GetRange()
+	return v
+}
+
+// GetEndKey returns the end key of the shard.
+func (r *CachedShard) GetEndKey() []byte {
+	_, v := r.Meta.GetRange()
+	return v
+}
+
+func (r *CachedShard) getLeaderID() uint64 {
+	if r.leader == nil {
+		return 0
+	}
+
+	return r.leader.ID
+}
+
+// shardMap wraps a map[uint64]*CachedShard and supports randomly pick a shard.
+type shardMap struct {
+	m         map[uint64]*CachedShard
+	totalSize int64
+	totalKeys int64
+}
+
+func newShardMap() *shardMap {
+	return &shardMap{
+		m: make(map[uint64]*CachedShard),
+	}
+}
+
+func (rm *shardMap) Len() int {
+	if rm == nil {
+		return 0
+	}
+	return len(rm.m)
+}
+
+func (rm *shardMap) Get(id uint64) *CachedShard {
+	if rm == nil {
+		return nil
+	}
+	if r, ok := rm.m[id]; ok {
+		return r
+	}
+	return nil
+}
+
+func (rm *shardMap) Put(res *CachedShard) {
+	if old, ok := rm.m[res.Meta.GetID()]; ok {
+		rm.totalSize -= int64(old.stats.ApproximateSize)
+		rm.totalKeys -= int64(old.stats.ApproximateKeys)
+	}
+	rm.m[res.Meta.GetID()] = res
+	rm.totalSize += int64(res.stats.ApproximateSize)
+	rm.totalKeys += int64(res.stats.ApproximateKeys)
+}
+
+func (rm *shardMap) Delete(id uint64) {
+	if rm == nil {
+		return
+	}
+	if old, ok := rm.m[id]; ok {
+		delete(rm.m, id)
+		rm.totalSize -= int64(old.stats.ApproximateSize)
+		rm.totalKeys -= int64(old.stats.ApproximateKeys)
+	}
+}
+
+func (rm *shardMap) TotalSize() int64 {
+	if rm.Len() == 0 {
+		return 0
+	}
+	return rm.totalSize
+}
+
+// shardSubTree is used to manager different types of shards.
+type shardSubTree struct {
+	*shardTree
+	totalSize int64
+	totalKeys int64
+}
+
+func newShardSubTree() *shardSubTree {
+	return &shardSubTree{
+		shardTree: newShardTree(),
+		totalSize: 0,
+	}
+}
+
+func (rst *shardSubTree) TotalSize() int64 {
+	if rst.length() == 0 {
+		return 0
+	}
+	return rst.totalSize
+}
+
+func (rst *shardSubTree) scanRanges() []*CachedShard {
+	if rst.length() == 0 {
+		return nil
+	}
+	var shards []*CachedShard
+	rst.scanRange([]byte(""), func(shard *CachedShard) bool {
+		shards = append(shards, shard)
+		return true
+	})
+	return shards
+}
+
+func (rst *shardSubTree) update(res *CachedShard) {
+	overlaps := rst.shardTree.update(res)
+	rst.totalSize += int64(res.stats.ApproximateSize)
+	rst.totalKeys += int64(res.stats.ApproximateKeys)
+	for _, r := range overlaps {
+		rst.totalSize -= int64(r.stats.ApproximateSize)
+		rst.totalKeys -= int64(r.stats.ApproximateKeys)
+	}
+}
+
+func (rst *shardSubTree) remove(res *CachedShard) {
+	if rst.length() == 0 {
+		return
+	}
+	if rst.shardTree.remove(res) != nil {
+		rst.totalSize -= int64(res.stats.ApproximateSize)
+		rst.totalKeys -= int64(res.stats.ApproximateKeys)
+	}
+}
+
+func (rst *shardSubTree) length() int {
+	if rst == nil {
+		return 0
+	}
+	return rst.shardTree.length()
+}
+
+func (rst *shardSubTree) RandomShards(n int, ranges []KeyRange) []*CachedShard {
+	if rst.length() == 0 {
+		return nil
+	}
+
+	shards := make([]*CachedShard, 0, n)
+
+	for i := 0; i < n; i++ {
+		if shard := rst.shardTree.RandomShard(ranges); shard != nil {
+			shards = append(shards, shard)
+		}
+	}
+	return shards
+}
+
+// ShardsContainer for export
+type ShardsContainer struct {
+	sync.RWMutex
+
+	trees           map[uint64]*shardTree               // group id -> shardTree
+	shards          *shardMap                           // shardID -> CachedShard
+	leaders         map[string]map[uint64]*shardSubTree // groupKey -> storeID -> shardSubTree
+	followers       map[string]map[uint64]*shardSubTree // groupKey -> storeID -> shardSubTree
+	learners        map[string]map[uint64]*shardSubTree // groupKey -> storeID -> shardSubTree
+	pendingReplicas map[string]map[uint64]*shardSubTree // groupKey -> storeID -> shardSubTree
+}
+
+// NewCachedShards creates ShardsContainer with tree, shards, leaders and followers
+func NewCachedShards() *ShardsContainer {
+	return &ShardsContainer{
+		trees:           make(map[uint64]*shardTree),
+		shards:          newShardMap(),
+		leaders:         make(map[string]map[uint64]*shardSubTree),
+		followers:       make(map[string]map[uint64]*shardSubTree),
+		learners:        make(map[string]map[uint64]*shardSubTree),
+		pendingReplicas: make(map[string]map[uint64]*shardSubTree),
+	}
+}
+
+func (r *ShardsContainer) maybeInitWithGroup(groupKey string) {
+	r.Lock()
+	defer r.Unlock()
+
+	if _, ok := r.leaders[groupKey]; !ok {
+		r.leaders[groupKey] = make(map[uint64]*shardSubTree)
+	}
+	if _, ok := r.followers[groupKey]; !ok {
+		r.followers[groupKey] = make(map[uint64]*shardSubTree)
+	}
+	if _, ok := r.learners[groupKey]; !ok {
+		r.learners[groupKey] = make(map[uint64]*shardSubTree)
+	}
+	if _, ok := r.pendingReplicas[groupKey]; !ok {
+		r.pendingReplicas[groupKey] = make(map[uint64]*shardSubTree)
+	}
+}
+
+// ForeachShards foreach shard by group
+func (r *ShardsContainer) ForeachShards(group uint64, fn func(res metapb.Shard)) {
+	for _, res := range r.shards.m {
+		if res.Meta.GetGroup() == group {
+			fn(res.Meta)
+		}
+	}
+}
+
+// ForeachCachedShards foreach cached shard by group
+func (r *ShardsContainer) ForeachCachedShards(group uint64, fn func(res *CachedShard)) {
+	for _, res := range r.shards.m {
+		if res.Meta.GetGroup() == group {
+			fn(res)
+		}
+	}
+}
+
+// GetShard returns the CachedShard with shardID
+func (r *ShardsContainer) GetShard(shardID uint64) *CachedShard {
+	return r.shards.Get(shardID)
+}
+
+// SetShard sets the CachedShard with shardID
+func (r *ShardsContainer) SetShard(res *CachedShard) []*CachedShard {
+	if origin := r.shards.Get(res.Meta.GetID()); origin != nil {
+		if !bytes.Equal(origin.GetStartKey(), res.GetStartKey()) ||
+			!bytes.Equal(origin.GetEndKey(), res.GetEndKey()) {
+			r.removeShardFromTreeAndMap(origin)
+		}
+		if r.shouldRemoveFromSubTree(res, origin) {
+			r.removeShardFromSubTree(origin)
+		}
+	}
+	return r.AddShard(res)
+}
+
+// Length returns the shardsInfo length
+func (r *ShardsContainer) Length() int {
+	return r.shards.Len()
+}
+
+// GetOverlaps returns the shards which are overlapped with the specified shard range.
+func (r *ShardsContainer) GetOverlaps(res *CachedShard) []*CachedShard {
+	if tree, ok := r.trees[res.Meta.GetGroup()]; ok {
+		return tree.getOverlaps(res)
+	}
+
+	return nil
+}
+
+// AddShard adds CachedShard to shardTree and shardMap, also update leaders and followers by shard peers
+func (r *ShardsContainer) AddShard(res *CachedShard) []*CachedShard {
+	if _, ok := r.trees[res.Meta.GetGroup()]; !ok {
+		r.trees[res.Meta.GetGroup()] = newShardTree()
+	}
+
+	// Destroying shard cannot add to tree to avoid range overlaps,
+	// and only wait schedule remove down replicas.
+	if res.IsDestroyState() {
+		r.RemoveShard(res)
+	}
+
+	tree := r.trees[res.Meta.GetGroup()]
+	// the shards which are overlapped with the specified shard range.
+	var overlaps []*CachedShard
+	// when the value is true, add the shard to the tree.
+	// Otherwise, use the shard replace the origin shard in the tree.
+	treeNeedAdd := !res.IsDestroyState()
+	if origin := r.GetShard(res.Meta.GetID()); origin != nil {
+		if resOld := tree.find(res); resOld != nil {
+			// Update to tree.
+			if bytes.Equal(resOld.shard.GetStartKey(), res.GetStartKey()) &&
+				bytes.Equal(resOld.shard.GetEndKey(), res.GetEndKey()) &&
+				resOld.shard.Meta.GetID() == res.Meta.GetID() {
+				resOld.shard = res
+				treeNeedAdd = false
+			}
+		}
+	}
+	if treeNeedAdd {
+		// Add to tree.
+		overlaps = tree.update(res)
+		for _, item := range overlaps {
+			r.RemoveShard(r.GetShard(item.Meta.GetID()))
+		}
+	}
+	// Add to shards.
+	r.shards.Put(res)
+
+	r.maybeInitWithGroup(res.groupKey)
+
+	// Add to leaders and followers.
+	for _, peer := range res.GetVoters() {
+		storeID := peer.StoreID
+		if peer.ID == res.getLeaderID() {
+			// Add leader peer to leaders.
+			store, ok := r.leaders[res.groupKey][storeID]
+			if !ok {
+				store = newShardSubTree()
+				r.leaders[res.groupKey][storeID] = store
+			}
+			store.update(res)
+		} else {
+			// Add follower peer to followers.
+			store, ok := r.followers[res.groupKey][storeID]
+			if !ok {
+				store = newShardSubTree()
+				r.followers[res.groupKey][storeID] = store
+			}
+			store.update(res)
+		}
+	}
+
+	// Add to learners.
+	for _, peer := range res.GetLearners() {
+		storeID := peer.StoreID
+		store, ok := r.learners[res.groupKey][storeID]
+		if !ok {
+			store = newShardSubTree()
+			r.learners[res.groupKey][storeID] = store
+		}
+		store.update(res)
+	}
+
+	for _, peer := range res.pendingReplicas {
+		storeID := peer.StoreID
+		store, ok := r.pendingReplicas[res.groupKey][storeID]
+		if !ok {
+			store = newShardSubTree()
+			r.pendingReplicas[res.groupKey][storeID] = store
+		}
+		store.update(res)
+	}
+
+	return overlaps
+}
+
+// RemoveShard removes CachedShard from shardTree and shardMap
+func (r *ShardsContainer) RemoveShard(res *CachedShard) {
+	// Remove from tree and shards.
+	r.removeShardFromTreeAndMap(res)
+	// Remove from leaders and followers.
+	r.removeShardFromSubTree(res)
+}
+
+// removeShardFromTreeAndMap removes CachedShard from shardTree and shardMap
+func (r *ShardsContainer) removeShardFromTreeAndMap(res *CachedShard) {
+	// Remove from tree and shards.
+	if tree, ok := r.trees[res.Meta.GetGroup()]; ok {
+		tree.remove(res)
+	}
+	r.shards.Delete(res.Meta.GetID())
+}
+
+// removeShardFromSubTree removes CachedShard from shardSubTrees
+func (r *ShardsContainer) removeShardFromSubTree(res *CachedShard) {
+	r.maybeInitWithGroup(res.groupKey)
+
+	// Remove from leaders and followers.
+	for _, peer := range res.Meta.GetReplicas() {
+		storeID := peer.StoreID
+		r.leaders[res.groupKey][storeID].remove(res)
+		r.followers[res.groupKey][storeID].remove(res)
+		r.learners[res.groupKey][storeID].remove(res)
+		r.pendingReplicas[res.groupKey][storeID].remove(res)
+	}
+}
+
+type replicaSlice []metapb.Replica
+
+func (s replicaSlice) Len() int {
+	return len(s)
+}
+func (s replicaSlice) Swap(i, j int) {
+	s[i], s[j] = s[j], s[i]
+}
+func (s replicaSlice) Less(i, j int) bool {
+	return s[i].ID < s[j].ID
+}
+
+// SortedPeersEqual judges whether two sorted `replicaSlice` are equal
+func SortedPeersEqual(peersA, peersB []metapb.Replica) bool {
+	if len(peersA) != len(peersB) {
+		return false
+	}
+	for i, peer := range peersA {
+		if peer.GetID() != peersB[i].GetID() {
+			return false
+		}
+	}
+	return true
+}
+
+type replicaStatsSlice []metapb.ReplicaStats
+
+func (s replicaStatsSlice) Len() int {
+	return len(s)
+}
+func (s replicaStatsSlice) Swap(i, j int) {
+	s[i], s[j] = s[j], s[i]
+}
+func (s replicaStatsSlice) Less(i, j int) bool {
+	return s[i].GetReplica().ID < s[j].GetReplica().ID
+}
+
+// SortedPeersStatsEqual judges whether two sorted `replicaStatsSlice` are equal
+func SortedPeersStatsEqual(peersA, peersB []metapb.ReplicaStats) bool {
+	if len(peersA) != len(peersB) {
+		return false
+	}
+	for i, peerStats := range peersA {
+		if peerStats.GetReplica().ID != peersB[i].GetReplica().ID ||
+			peerStats.GetDownSeconds() != peersB[i].GetDownSeconds() {
+			return false
+		}
+	}
+	return true
+}
+
+// shouldRemoveFromSubTree return true when the shard leader changed, peer transferred,
+// new peer was created, learners changed, pendingReplicas changed, and so on.
+func (r *ShardsContainer) shouldRemoveFromSubTree(res *CachedShard, origin *CachedShard) bool {
+	checkPeersChange := func(origin []metapb.Replica, other []metapb.Replica) bool {
+		if len(origin) != len(other) {
+			return true
+		}
+		sort.Sort(replicaSlice(origin))
+		sort.Sort(replicaSlice(other))
+		for index, peer := range origin {
+			if peer.StoreID == other[index].StoreID && peer.ID == other[index].ID {
+				continue
+			}
+			return true
+		}
+		return false
+	}
+
+	return origin.getLeaderID() != res.getLeaderID() ||
+		checkPeersChange(origin.GetVoters(), res.GetVoters()) ||
+		checkPeersChange(origin.GetLearners(), res.GetLearners()) ||
+		checkPeersChange(origin.GetPendingPeers(), res.GetPendingPeers()) ||
+		origin.groupKey != res.groupKey
+}
+
+// SearchShard searches CachedShard from shardTree
+func (r *ShardsContainer) SearchShard(group uint64, resKey []byte) *CachedShard {
+	if tree, ok := r.trees[group]; ok {
+		res := tree.search(resKey)
+		if res == nil {
+			return nil
+		}
+		return r.GetShard(res.Meta.GetID())
+	}
+
+	return nil
+}
+
+// SearchPrevShard searches previous CachedShard from shardTree
+func (r *ShardsContainer) SearchPrevShard(group uint64, resKey []byte) *CachedShard {
+	if tree, ok := r.trees[group]; ok {
+		res := tree.searchPrev(resKey)
+		if res == nil {
+			return nil
+		}
+		return r.GetShard(res.Meta.GetID())
+	}
+
+	return nil
+}
+
+// GetShards gets all CachedShard from shardMap
+func (r *ShardsContainer) GetShards() []*CachedShard {
+	shards := make([]*CachedShard, 0, r.shards.Len())
+	for _, res := range r.shards.m {
+		shards = append(shards, res)
+	}
+	return shards
+}
+
+// GetStoreShards gets all CachedShard with a given storeID
+func (r *ShardsContainer) GetStoreShards(groupKey string, storeID uint64) []*CachedShard {
+	r.maybeInitWithGroup(groupKey)
+	shards := make([]*CachedShard, 0, r.GetStoreShardCount(groupKey, storeID))
+	if leaders, ok := r.leaders[groupKey][storeID]; ok {
+		shards = append(shards, leaders.scanRanges()...)
+	}
+	if followers, ok := r.followers[groupKey][storeID]; ok {
+		shards = append(shards, followers.scanRanges()...)
+	}
+	if learners, ok := r.learners[groupKey][storeID]; ok {
+		shards = append(shards, learners.scanRanges()...)
+	}
+	return shards
+}
+
+// GetStoreLeaderShardSize get total size of container's leader shards
+func (r *ShardsContainer) GetStoreLeaderShardSize(groupKey string, storeID uint64) int64 {
+	r.maybeInitWithGroup(groupKey)
+	return r.leaders[groupKey][storeID].TotalSize()
+}
+
+// GetStoreFollowerShardSize get total size of container's follower shards
+func (r *ShardsContainer) GetStoreFollowerShardSize(groupKey string, storeID uint64) int64 {
+	r.maybeInitWithGroup(groupKey)
+	return r.followers[groupKey][storeID].TotalSize()
+}
+
+// GetStoreLearnerShardSize get total size of container's learner shards
+func (r *ShardsContainer) GetStoreLearnerShardSize(groupKey string, storeID uint64) int64 {
+	r.maybeInitWithGroup(groupKey)
+	return r.learners[groupKey][storeID].TotalSize()
+}
+
+// GetStoreShardSize get total size of container's shards
+func (r *ShardsContainer) GetStoreShardSize(groupKey string, storeID uint64) int64 {
+	r.maybeInitWithGroup(groupKey)
+	return r.GetStoreLeaderShardSize(groupKey, storeID) +
+		r.GetStoreFollowerShardSize(groupKey, storeID) +
+		r.GetStoreLearnerShardSize(groupKey, storeID)
+}
+
+// GetMetaShards gets a set of metapb.Shard from shardMap
+func (r *ShardsContainer) GetMetaShards() []metapb.Shard {
+	shards := make([]metapb.Shard, 0, r.shards.Len())
+	for _, res := range r.shards.m {
+		shards = append(shards, res.Meta)
+	}
+	return shards
+}
+
+// GetShardCount gets the total count of CachedShard of shardMap
+func (r *ShardsContainer) GetShardCount() int {
+	return r.shards.Len()
+}
+
+// GetStoreShardCount gets the total count of a container's leader, follower and learner CachedShard by storeID
+func (r *ShardsContainer) GetStoreShardCount(groupKey string, storeID uint64) int {
+	r.maybeInitWithGroup(groupKey)
+	return r.GetStoreLeaderCount(groupKey, storeID) +
+		r.GetStoreFollowerCount(groupKey, storeID) +
+		r.GetStoreLearnerCount(groupKey, storeID)
+}
+
+// GetStorePendingPeerCount gets the total count of a container's shard that includes pending peer
+func (r *ShardsContainer) GetStorePendingPeerCount(groupKey string, storeID uint64) int {
+	r.maybeInitWithGroup(groupKey)
+	return r.pendingReplicas[groupKey][storeID].length()
+}
+
+// GetStoreLeaderCount get the total count of a container's leader CachedShard
+func (r *ShardsContainer) GetStoreLeaderCount(groupKey string, storeID uint64) int {
+	r.maybeInitWithGroup(groupKey)
+	return r.leaders[groupKey][storeID].length()
+}
+
+// GetStoreFollowerCount get the total count of a container's follower CachedShard
+func (r *ShardsContainer) GetStoreFollowerCount(groupKey string, storeID uint64) int {
+	r.maybeInitWithGroup(groupKey)
+	return r.followers[groupKey][storeID].length()
+}
+
+// GetStoreLearnerCount get the total count of a container's learner CachedShard
+func (r *ShardsContainer) GetStoreLearnerCount(groupKey string, storeID uint64) int {
+	r.maybeInitWithGroup(groupKey)
+	return r.learners[groupKey][storeID].length()
+}
+
+// RandPendingShards randomly gets a container's n shards with a pending peer.
+func (r *ShardsContainer) RandPendingShards(groupKey string, storeID uint64, ranges []KeyRange, n int) []*CachedShard {
+	r.maybeInitWithGroup(groupKey)
+	return r.pendingReplicas[groupKey][storeID].RandomShards(n, ranges)
+}
+
+// RandLeaderShards randomly gets a container's n leader shards.
+func (r *ShardsContainer) RandLeaderShards(groupKey string, storeID uint64, ranges []KeyRange, n int) []*CachedShard {
+	r.maybeInitWithGroup(groupKey)
+	return r.leaders[groupKey][storeID].RandomShards(n, ranges)
+}
+
+// RandFollowerShards randomly gets a container's n follower shards.
+func (r *ShardsContainer) RandFollowerShards(groupKey string, storeID uint64, ranges []KeyRange, n int) []*CachedShard {
+	r.maybeInitWithGroup(groupKey)
+	return r.followers[groupKey][storeID].RandomShards(n, ranges)
+}
+
+// RandLearnerShards randomly gets a container's n learner shards.
+func (r *ShardsContainer) RandLearnerShards(groupKey string, storeID uint64, ranges []KeyRange, n int) []*CachedShard {
+	r.maybeInitWithGroup(groupKey)
+	return r.learners[groupKey][storeID].RandomShards(n, ranges)
+}
+
+// GetLeader return leader CachedShard by storeID and shardID(now only used in test)
+func (r *ShardsContainer) GetLeader(groupKey string, storeID uint64, res *CachedShard) *CachedShard {
+	r.maybeInitWithGroup(groupKey)
+	if leaders, ok := r.leaders[groupKey][storeID]; ok {
+		return leaders.find(res).shard
+	}
+	return nil
+}
+
+// GetFollower return follower CachedShard by storeID and shardID(now only used in test)
+func (r *ShardsContainer) GetFollower(groupKey string, storeID uint64, res *CachedShard) *CachedShard {
+	r.maybeInitWithGroup(groupKey)
+	if followers, ok := r.followers[groupKey][storeID]; ok {
+		return followers.find(res).shard
+	}
+	return nil
+}
+
+// ScanRange scans shards intersecting [start key, end key), returns at most
+// `limit` shards. limit <= 0 means no limit.
+func (r *ShardsContainer) ScanRange(group uint64, startKey, endKey []byte, limit int) []*CachedShard {
+	var shards []*CachedShard
+	if tree, ok := r.trees[group]; ok {
+		tree.scanRange(startKey, func(shard *CachedShard) bool {
+			if len(endKey) > 0 && bytes.Compare(shard.GetStartKey(), endKey) >= 0 {
+				return false
+			}
+			if limit > 0 && len(shards) >= limit {
+				return false
+			}
+			shards = append(shards, r.GetShard(shard.Meta.GetID()))
+			return true
+		})
+	}
+	return shards
+}
+
+// GetDestroyingShards returns all shards in destroying state
+func (r *ShardsContainer) GetDestroyingShards() []*CachedShard {
+	var shards []*CachedShard
+	for _, res := range r.shards.m {
+		if res.Meta.GetState() == metapb.ShardState_Destroying {
+			shards = append(shards, res)
+		}
+	}
+	return shards
+}
+
+// ScanRangeWithIterator scans from the first shard containing or behind start key,
+// until iterator returns false.
+func (r *ShardsContainer) ScanRangeWithIterator(group uint64, startKey []byte, iterator func(res *CachedShard) bool) {
+	if tree, ok := r.trees[group]; ok {
+		tree.scanRange(startKey, iterator)
+	}
+}
+
+// GetAdjacentShards returns shard's info that is adjacent with specific shard
+func (r *ShardsContainer) GetAdjacentShards(res *CachedShard) (*CachedShard, *CachedShard) {
+	var prev, next *CachedShard
+	if tree, ok := r.trees[res.Meta.GetGroup()]; ok {
+		p, n := tree.getAdjacentShards(res)
+		// check key to avoid key range hole
+		if p != nil && bytes.Equal(p.shard.GetEndKey(), res.GetStartKey()) {
+			prev = r.GetShard(p.shard.Meta.GetID())
+		}
+		if n != nil && bytes.Equal(res.GetEndKey(), n.shard.GetStartKey()) {
+			next = r.GetShard(n.shard.Meta.GetID())
+		}
+	}
+	return prev, next
+}
+
+// GetAverageShardSize returns the average shard approximate size.
+func (r *ShardsContainer) GetAverageShardSize() int64 {
+	if r.shards.Len() == 0 {
+		return 0
+	}
+	return r.shards.TotalSize() / int64(r.shards.Len())
+}
+
+// DiffShardPeersInfo return the difference of peers info  between two CachedShard
+func DiffShardPeersInfo(origin *CachedShard, other *CachedShard) string {
+	var ret []string
+	for _, a := range origin.Meta.GetReplicas() {
+		both := false
+		for _, b := range other.Meta.GetReplicas() {
+			if reflect.DeepEqual(a, b) {
+				both = true
+				break
+			}
+		}
+		if !both {
+			ret = append(ret, fmt.Sprintf("Remove peer:{%v}", a))
+		}
+	}
+	for _, b := range other.Meta.GetReplicas() {
+		both := false
+		for _, a := range origin.Meta.GetReplicas() {
+			if reflect.DeepEqual(a, b) {
+				both = true
+				break
+			}
+		}
+		if !both {
+			ret = append(ret, fmt.Sprintf("Add peer:{%v}", b))
+		}
+	}
+	return strings.Join(ret, ",")
+}
+
+// DiffShardKeyInfo return the difference of key info between two CachedShard
+func DiffShardKeyInfo(origin *CachedShard, other *CachedShard) string {
+	originStartKey, originEndKey := origin.Meta.GetRange()
+	otherStartKey, otherEndKey := other.Meta.GetRange()
+
+	var ret []string
+	if !bytes.Equal(originStartKey, otherStartKey) {
+		ret = append(ret, fmt.Sprintf("StartKey Changed:{%s} -> {%s}", HexShardKey(originStartKey), HexShardKey(otherStartKey)))
+	} else {
+		ret = append(ret, fmt.Sprintf("StartKey:{%s}", HexShardKey(originStartKey)))
+	}
+	if !bytes.Equal(originEndKey, otherEndKey) {
+		ret = append(ret, fmt.Sprintf("EndKey Changed:{%s} -> {%s}", HexShardKey(originEndKey), HexShardKey(otherEndKey)))
+	} else {
+		ret = append(ret, fmt.Sprintf("EndKey:{%s}", HexShardKey(originEndKey)))
+	}
+
+	return strings.Join(ret, ", ")
+}
+
+func isInvolved(res *CachedShard, startKey, endKey []byte) bool {
+	return bytes.Compare(res.GetStartKey(), startKey) >= 0 && (len(endKey) == 0 || (len(res.GetEndKey()) > 0 && bytes.Compare(res.GetEndKey(), endKey) <= 0))
+}
+
+// String converts slice of bytes to string without copy.
+func String(b []byte) (s string) {
+	if len(b) == 0 {
+		return ""
+	}
+	pbytes := (*reflect.SliceHeader)(unsafe.Pointer(&b))
+	pstring := (*reflect.StringHeader)(unsafe.Pointer(&s))
+	pstring.Data = pbytes.Data
+	pstring.Len = pbytes.Len
+	return
+}
+
+// ToUpperASCIIInplace bytes.ToUpper but zero-cost
+func ToUpperASCIIInplace(s []byte) []byte {
+	hasLower := false
+	for i := 0; i < len(s); i++ {
+		c := s[i]
+		hasLower = hasLower || ('a' <= c && c <= 'z')
+	}
+
+	if !hasLower {
+		return s
+	}
+	var c byte
+	for i := 0; i < len(s); i++ {
+		c = s[i]
+		if 'a' <= c && c <= 'z' {
+			c -= 'a' - 'A'
+		}
+		s[i] = c
+	}
+	return s
+}
+
+// EncodeToString overrides hex.EncodeToString implementation. Difference: returns []byte, not string
+func EncodeToString(src []byte) []byte {
+	dst := make([]byte, hex.EncodedLen(len(src)))
+	hex.Encode(dst, src)
+	return dst
+}
+
+// HexShardKey converts shard key to hex format. Used for formatting shard in
+// logs.
+func HexShardKey(key []byte) []byte {
+	return ToUpperASCIIInplace(EncodeToString(key))
+}
+
+// ShardToHexMeta converts a shard meta's keys to hex format. Used for formatting
+// shard in logs.
+func ShardToHexMeta(meta metapb.Shard) HexShardMeta {
+	start, end := meta.GetRange()
+	meta.SetStartKey(HexShardKey(start))
+	meta.SetEndKey(HexShardKey(end))
+	return HexShardMeta{meta}
+}
+
+// HexShardMeta is a shard meta in the hex format. Used for formatting shard in logs.
+type HexShardMeta struct {
+	meta metapb.Shard
+}
+
+func (h HexShardMeta) String() string {
+	return fmt.Sprintf("shard %+v", h.meta)
+}
diff --git a/components/prophet/core/resource_option.go b/components/prophet/core/shard_option.go
similarity index 80%
rename from components/prophet/core/resource_option.go
rename to components/prophet/core/shard_option.go
index cf4102c..5cd198e 100644
--- a/components/prophet/core/resource_option.go
+++ b/components/prophet/core/shard_option.go
@@ -38,7 +38,7 @@ func WithState(state metapb.ShardState) ShardCreateOption {
 func WithDownPeers(downReplicas []metapb.ReplicaStats) ShardCreateOption {
 	return func(res *CachedShard) {
 		res.downReplicas = append(downReplicas[:0:0], downReplicas...)
-		sort.Sort(peerStatsSlice(res.downReplicas))
+		sort.Sort(res.downReplicas)
 	}
 }
 
@@ -46,7 +46,7 @@ func WithDownPeers(downReplicas []metapb.ReplicaStats) ShardCreateOption {
 func WithPendingPeers(pendingReplicas []metapb.Replica) ShardCreateOption {
 	return func(res *CachedShard) {
 		res.pendingReplicas = append(pendingReplicas[:0:0], pendingReplicas...)
-		sort.Sort(peerSlice(res.pendingReplicas))
+		sort.Sort(res.pendingReplicas)
 	}
 }
 
@@ -93,20 +93,6 @@ func WithNewShardID(id uint64) ShardCreateOption {
 	}
 }
 
-// WithNewPeerIds sets new ids for peers.
-func WithNewPeerIds(peerIDs ...uint64) ShardCreateOption {
-	return func(res *CachedShard) {
-		if len(peerIDs) != len(res.Meta.GetReplicas()) {
-			return
-		}
-
-		peers := res.Meta.GetReplicas()
-		for i := range peers {
-			peers[i].ID = peerIDs[i]
-		}
-	}
-}
-
 // WithIncVersion increases the version of the resource.
 func WithIncVersion() ShardCreateOption {
 	return func(res *CachedShard) {
@@ -134,15 +120,6 @@ func WithIncConfVer() ShardCreateOption {
 	}
 }
 
-// WithDecConfVer decreases the config version of the resource.
-func WithDecConfVer() ShardCreateOption {
-	return func(res *CachedShard) {
-		e := res.Meta.GetEpoch()
-		e.ConfigVer--
-		res.Meta.SetEpoch(e)
-	}
-}
-
 // SetWrittenBytes sets the written bytes for the resource.
 func SetWrittenBytes(v uint64) ShardCreateOption {
 	return func(res *CachedShard) {
@@ -205,32 +182,6 @@ func SetReportInterval(v uint64) ShardCreateOption {
 	}
 }
 
-// SetShardConfVer sets the config version for the resource.
-func SetShardConfVer(confVer uint64) ShardCreateOption {
-	return func(res *CachedShard) {
-		e := res.Meta.GetEpoch()
-		if e.Generation == 0 {
-			res.Meta.SetEpoch(metapb.ShardEpoch{ConfigVer: confVer, Generation: 1})
-		} else {
-			e.ConfigVer = confVer
-			res.Meta.SetEpoch(e)
-		}
-	}
-}
-
-// SetShardVersion sets the version for the resource.
-func SetShardVersion(version uint64) ShardCreateOption {
-	return func(res *CachedShard) {
-		e := res.Meta.GetEpoch()
-		if e.Generation == 0 {
-			res.Meta.SetEpoch(metapb.ShardEpoch{ConfigVer: 1, Generation: version})
-		} else {
-			e.Generation = version
-			res.Meta.SetEpoch(e)
-		}
-	}
-}
-
 // SetPeers sets the peers for the resource.
 func SetPeers(peers []metapb.Replica) ShardCreateOption {
 	return func(res *CachedShard) {
@@ -277,10 +228,3 @@ func WithReplacePeerStore(oldStoreID, newStoreID uint64) ShardCreateOption {
 		}
 	}
 }
-
-// WithInterval sets the interval
-func WithInterval(interval *metapb.TimeInterval) ShardCreateOption {
-	return func(res *CachedShard) {
-		res.stats.Interval = interval
-	}
-}
diff --git a/components/prophet/core/resource_tree.go b/components/prophet/core/shard_tree.go
similarity index 59%
rename from components/prophet/core/resource_tree.go
rename to components/prophet/core/shard_tree.go
index b29530c..4d14f12 100644
--- a/components/prophet/core/resource_tree.go
+++ b/components/prophet/core/shard_tree.go
@@ -16,28 +16,28 @@ package core
 
 import (
 	"bytes"
-	"github.com/matrixorigin/matrixcube/pb/metapb"
 	"math/rand"
 	"time"
 
 	"github.com/matrixorigin/matrixcube/components/prophet/util/btree"
+	"github.com/matrixorigin/matrixcube/pb/metapb"
 )
 
-var _ btree.Item = &resourceItem{}
+var _ btree.Item = &shardItem{}
 
-type resourceItem struct {
-	res *CachedShard
+type shardItem struct {
+	shard *CachedShard
 }
 
 // Less returns true if the resource start key is less than the other.
-func (r *resourceItem) Less(other btree.Item) bool {
-	left := r.res.GetStartKey()
-	right := other.(*resourceItem).res.GetStartKey()
+func (r *shardItem) Less(other btree.Item) bool {
+	left := r.shard.GetStartKey()
+	right := other.(*shardItem).shard.GetStartKey()
 	return bytes.Compare(left, right) < 0
 }
 
-func (r *resourceItem) Contains(key []byte) bool {
-	start, end := r.res.GetStartKey(), r.res.GetEndKey()
+func (r *shardItem) Contains(key []byte) bool {
+	start, end := r.shard.GetStartKey(), r.shard.GetEndKey()
 	return bytes.Compare(key, start) >= 0 && (len(end) == 0 || bytes.Compare(key, end) < 0)
 }
 
@@ -45,36 +45,34 @@ const (
 	defaultBTreeDegree = 64
 )
 
-type resourceTree struct {
+type shardTree struct {
 	tree *btree.BTree
 }
 
-func newShardTree() *resourceTree {
-	return &resourceTree{
+func newShardTree() *shardTree {
+	return &shardTree{
 		tree: btree.New(defaultBTreeDegree),
 	}
 }
 
-func (t *resourceTree) newSearchRes(key []byte) *CachedShard {
-	meta := metapb.Shard{}
-	meta.SetStartKey(key)
-	return &CachedShard{Meta: meta}
+func (t *shardTree) newSearchRes(key []byte) *CachedShard {
+	return &CachedShard{Meta: metapb.Shard{Start: key}}
 }
 
-func (t *resourceTree) length() int {
+func (t *shardTree) length() int {
 	return t.tree.Len()
 }
 
-// getOverlaps gets the resources which are overlapped with the specified resource range.
-func (t *resourceTree) getOverlaps(res *CachedShard) []*CachedShard {
-	item := &resourceItem{res: res}
+// getOverlaps gets the shards which are overlapped with the specified resource range.
+func (t *shardTree) getOverlaps(res *CachedShard) []*CachedShard {
+	item := &shardItem{shard: res}
 
 	// note that find() gets the last item that is less or equal than the resource.
 	// in the case: |_______a_______|_____b_____|___c___|
 	// new resource is   |______d______|
-	// find() will return resourceItem of resource_a
+	// find() will return shardItem of resource_a
 	// and both startKey of resource_a and resource_b are less than endKey of resource_d,
-	// thus they are regarded as overlapped resources.
+	// thus they are regarded as overlapped shards.
 	result := t.find(res)
 	if result == nil {
 		result = item
@@ -82,26 +80,26 @@ func (t *resourceTree) getOverlaps(res *CachedShard) []*CachedShard {
 
 	var overlaps []*CachedShard
 	t.tree.AscendGreaterOrEqual(result, func(i btree.Item) bool {
-		over := i.(*resourceItem)
-		if len(res.GetEndKey()) > 0 && bytes.Compare(res.GetEndKey(), over.res.GetStartKey()) <= 0 {
+		over := i.(*shardItem)
+		if len(res.GetEndKey()) > 0 && bytes.Compare(res.GetEndKey(), over.shard.GetStartKey()) <= 0 {
 			return false
 		}
-		overlaps = append(overlaps, over.res)
+		overlaps = append(overlaps, over.shard)
 		return true
 	})
 	return overlaps
 }
 
 // update updates the tree with the resource.
-// It finds and deletes all the overlapped resources first, and then
+// It finds and deletes all the overlapped shards first, and then
 // insert the resource.
-func (t *resourceTree) update(res *CachedShard) []*CachedShard {
+func (t *shardTree) update(res *CachedShard) []*CachedShard {
 	overlaps := t.getOverlaps(res)
 	for _, item := range overlaps {
-		t.tree.Delete(&resourceItem{item})
+		t.tree.Delete(&shardItem{item})
 	}
 
-	t.tree.ReplaceOrInsert(&resourceItem{res: res})
+	t.tree.ReplaceOrInsert(&shardItem{shard: res})
 
 	return overlaps
 }
@@ -109,12 +107,12 @@ func (t *resourceTree) update(res *CachedShard) []*CachedShard {
 // remove removes a resource if the resource is in the tree.
 // It will do nothing if it cannot find the resource or the found resource
 // is not the same with the resource.
-func (t *resourceTree) remove(res *CachedShard) btree.Item {
+func (t *shardTree) remove(res *CachedShard) btree.Item {
 	if t.length() == 0 {
 		return nil
 	}
 	result := t.find(res)
-	if result == nil || result.res.Meta.GetID() != res.Meta.GetID() {
+	if result == nil || result.shard.Meta.GetID() != res.Meta.GetID() {
 		return nil
 	}
 
@@ -122,40 +120,40 @@ func (t *resourceTree) remove(res *CachedShard) btree.Item {
 }
 
 // search returns a resource that contains the key.
-func (t *resourceTree) search(resKey []byte) *CachedShard {
+func (t *shardTree) search(resKey []byte) *CachedShard {
 	res := t.newSearchRes(resKey)
 	result := t.find(res)
 	if result == nil {
 		return nil
 	}
-	return result.res
+	return result.shard
 }
 
 // searchPrev returns the previous resource of the resource where the resourceKey is located.
-func (t *resourceTree) searchPrev(resKey []byte) *CachedShard {
+func (t *shardTree) searchPrev(resKey []byte) *CachedShard {
 	curRes := t.newSearchRes(resKey)
 	curResItem := t.find(curRes)
 	if curResItem == nil {
 		return nil
 	}
-	prevShardItem, _ := t.getAdjacentShards(curResItem.res)
+	prevShardItem, _ := t.getAdjacentShards(curResItem.shard)
 	if prevShardItem == nil {
 		return nil
 	}
-	if !bytes.Equal(prevShardItem.res.GetEndKey(), curResItem.res.GetStartKey()) {
+	if !bytes.Equal(prevShardItem.shard.GetEndKey(), curResItem.shard.GetStartKey()) {
 		return nil
 	}
-	return prevShardItem.res
+	return prevShardItem.shard
 }
 
-// find is a helper function to find an item that contains the resources start
+// find is a helper function to find an item that contains the shards start
 // key.
-func (t *resourceTree) find(res *CachedShard) *resourceItem {
-	item := &resourceItem{res: res}
+func (t *shardTree) find(res *CachedShard) *shardItem {
+	item := &shardItem{shard: res}
 
-	var result *resourceItem
+	var result *shardItem
 	t.tree.DescendLessOrEqual(item, func(i btree.Item) bool {
-		result = i.(*resourceItem)
+		result = i.(*shardItem)
 		return false
 	})
 
@@ -168,40 +166,40 @@ func (t *resourceTree) find(res *CachedShard) *resourceItem {
 
 // scanRage scans from the first resource containing or behind the start key
 // until f return false
-func (t *resourceTree) scanRange(startKey []byte, f func(*CachedShard) bool) {
+func (t *shardTree) scanRange(startKey []byte, f func(*CachedShard) bool) {
 	res := t.newSearchRes(startKey)
 	// find if there is a resource with key range [s, d), s < startKey < d
 	startItem := t.find(res)
 	if startItem == nil {
-		startItem = &resourceItem{res: t.newSearchRes(startKey)}
+		startItem = &shardItem{shard: t.newSearchRes(startKey)}
 	}
 	t.tree.AscendGreaterOrEqual(startItem, func(item btree.Item) bool {
-		return f(item.(*resourceItem).res)
+		return f(item.(*shardItem).shard)
 	})
 }
 
-func (t *resourceTree) getAdjacentShards(res *CachedShard) (*resourceItem, *resourceItem) {
-	item := &resourceItem{res: t.newSearchRes(res.GetStartKey())}
-	var prev, next *resourceItem
+func (t *shardTree) getAdjacentShards(res *CachedShard) (*shardItem, *shardItem) {
+	item := &shardItem{shard: t.newSearchRes(res.GetStartKey())}
+	var prev, next *shardItem
 	t.tree.AscendGreaterOrEqual(item, func(i btree.Item) bool {
-		if bytes.Equal(item.res.GetStartKey(), i.(*resourceItem).res.GetStartKey()) {
+		if bytes.Equal(item.shard.GetStartKey(), i.(*shardItem).shard.GetStartKey()) {
 			return true
 		}
-		next = i.(*resourceItem)
+		next = i.(*shardItem)
 		return false
 	})
 	t.tree.DescendLessOrEqual(item, func(i btree.Item) bool {
-		if bytes.Equal(item.res.GetStartKey(), i.(*resourceItem).res.GetStartKey()) {
+		if bytes.Equal(item.shard.GetStartKey(), i.(*shardItem).shard.GetStartKey()) {
 			return true
 		}
-		prev = i.(*resourceItem)
+		prev = i.(*shardItem)
 		return false
 	})
 	return prev, next
 }
 
 // RandomShard is used to get a random resource within ranges.
-func (t *resourceTree) RandomShard(ranges []KeyRange) *CachedShard {
+func (t *shardTree) RandomShard(ranges []KeyRange) *CachedShard {
 	if t.length() == 0 {
 		return nil
 	}
@@ -213,17 +211,17 @@ func (t *resourceTree) RandomShard(ranges []KeyRange) *CachedShard {
 	for _, i := range rand.Perm(len(ranges)) {
 		var endIndex int
 		startKey, endKey := ranges[i].StartKey, ranges[i].EndKey
-		startShard, startIndex := t.tree.GetWithIndex(&resourceItem{res: t.newSearchRes(startKey)})
+		startShard, startIndex := t.tree.GetWithIndex(&shardItem{shard: t.newSearchRes(startKey)})
 
 		if len(endKey) != 0 {
-			_, endIndex = t.tree.GetWithIndex(&resourceItem{res: t.newSearchRes(endKey)})
+			_, endIndex = t.tree.GetWithIndex(&shardItem{shard: t.newSearchRes(endKey)})
 		} else {
 			endIndex = t.tree.Len()
 		}
 
 		// Consider that the item in the tree may not be continuous,
 		// we need to check if the previous item contains the key.
-		if startIndex != 0 && startShard == nil && t.tree.GetAt(startIndex-1).(*resourceItem).Contains(startKey) {
+		if startIndex != 0 && startShard == nil && t.tree.GetAt(startIndex-1).(*shardItem).Contains(startKey) {
 			startIndex--
 		}
 
@@ -231,7 +229,7 @@ func (t *resourceTree) RandomShard(ranges []KeyRange) *CachedShard {
 			continue
 		}
 		index := rand.Intn(endIndex-startIndex) + startIndex
-		res := t.tree.GetAt(index).(*resourceItem).res
+		res := t.tree.GetAt(index).(*shardItem).shard
 		if isInvolved(res, startKey, endKey) {
 			return res
 		}
diff --git a/components/prophet/core/container.go b/components/prophet/core/store.go
similarity index 60%
rename from components/prophet/core/container.go
rename to components/prophet/core/store.go
index 4248610..91f81ad 100755
--- a/components/prophet/core/container.go
+++ b/components/prophet/core/store.go
@@ -29,7 +29,6 @@ import (
 const (
 	// Interval to save container meta (including heartbeat ts) to etcd.
 	containerPersistInterval = 5 * time.Minute
-	mb                       = 1 << 20 // megabyte
 	gb                       = 1 << 30 // 1GB size
 	initialMaxShardCounts    = 30      // exclude storage Threshold Filter when resource less than 30
 	initialMinSpace          = 1 << 33 // 2^3=8GB
@@ -43,94 +42,94 @@ type counterAndSize struct {
 // CachedStore is the container runtime info cached in the cache
 type CachedStore struct {
 	Meta metapb.Store
-	*containerStats
+	*storeStats
 	pauseLeaderTransfer bool // not allow to be used as source or target of transfer leader
-	resourceInfo        map[string]counterAndSize
+	shardInfo           map[string]counterAndSize
 	leaderInfo          map[string]counterAndSize
 	pendingPeerCounts   map[string]int
 	lastPersistTime     time.Time
 	leaderWeight        float64
-	resourceWeight      float64
+	shardWeight         float64
 	available           map[limit.Type]func() bool
 }
 
-// NewCachedStore creates CachedStore with meta data.
+// NewCachedStore creates CachedStore with metadata.
 func NewCachedStore(meta metapb.Store, opts ...StoreCreateOption) *CachedStore {
-	container := &CachedStore{
+	store := &CachedStore{
 		Meta:              meta,
-		containerStats:    newStoreStats(),
-		resourceInfo:      make(map[string]counterAndSize),
+		storeStats:        newStoreStats(),
+		shardInfo:         make(map[string]counterAndSize),
 		leaderInfo:        make(map[string]counterAndSize),
 		pendingPeerCounts: make(map[string]int),
 		leaderWeight:      1.0,
-		resourceWeight:    1.0,
+		shardWeight:       1.0,
 	}
 	for _, opt := range opts {
-		opt(container)
+		opt(store)
 	}
-	return container
+	return store
 }
 
 // Clone creates a copy of current CachedStore.
 func (cr *CachedStore) Clone(opts ...StoreCreateOption) *CachedStore {
-	container := &CachedStore{
+	store := &CachedStore{
 		Meta:                cr.Meta.CloneValue(),
-		containerStats:      cr.containerStats,
+		storeStats:          cr.storeStats,
 		pauseLeaderTransfer: cr.pauseLeaderTransfer,
-		resourceInfo:        make(map[string]counterAndSize),
+		shardInfo:           make(map[string]counterAndSize),
 		leaderInfo:          make(map[string]counterAndSize),
 		pendingPeerCounts:   make(map[string]int),
 		lastPersistTime:     cr.lastPersistTime,
 		leaderWeight:        cr.leaderWeight,
-		resourceWeight:      cr.resourceWeight,
+		shardWeight:         cr.shardWeight,
 		available:           cr.available,
 	}
 
-	for k, v := range cr.resourceInfo {
-		container.resourceInfo[k] = v
+	for k, v := range cr.shardInfo {
+		store.shardInfo[k] = v
 	}
 	for k, v := range cr.leaderInfo {
-		container.leaderInfo[k] = v
+		store.leaderInfo[k] = v
 	}
 	for k, v := range cr.pendingPeerCounts {
-		container.pendingPeerCounts[k] = v
+		store.pendingPeerCounts[k] = v
 	}
 
 	for _, opt := range opts {
-		opt(container)
+		opt(store)
 	}
-	return container
+	return store
 }
 
 // ShallowClone creates a copy of current CachedStore, but not clone 'Meta'.
 func (cr *CachedStore) ShallowClone(opts ...StoreCreateOption) *CachedStore {
-	container := &CachedStore{
+	shard := &CachedStore{
 		Meta:                cr.Meta,
-		containerStats:      cr.containerStats,
+		storeStats:          cr.storeStats,
 		pauseLeaderTransfer: cr.pauseLeaderTransfer,
-		resourceInfo:        make(map[string]counterAndSize),
+		shardInfo:           make(map[string]counterAndSize),
 		leaderInfo:          make(map[string]counterAndSize),
 		pendingPeerCounts:   make(map[string]int),
 		lastPersistTime:     cr.lastPersistTime,
 		leaderWeight:        cr.leaderWeight,
-		resourceWeight:      cr.resourceWeight,
+		shardWeight:         cr.shardWeight,
 		available:           cr.available,
 	}
 
-	for k, v := range cr.resourceInfo {
-		container.resourceInfo[k] = v
+	for k, v := range cr.shardInfo {
+		shard.shardInfo[k] = v
 	}
 	for k, v := range cr.leaderInfo {
-		container.leaderInfo[k] = v
+		shard.leaderInfo[k] = v
 	}
 	for k, v := range cr.pendingPeerCounts {
-		container.pendingPeerCounts[k] = v
+		shard.pendingPeerCounts[k] = v
 	}
 
 	for _, opt := range opts {
-		opt(container)
+		opt(shard)
 	}
-	return container
+	return shard
 }
 
 // AllowLeaderTransfer returns if the container is allowed to be selected
@@ -193,22 +192,22 @@ func (cr *CachedStore) GetTotalLeaderCount() int {
 
 // GetShardCount returns the Shard count of the container.
 func (cr *CachedStore) GetShardCount(groupKey string) int {
-	return cr.resourceInfo[groupKey].count
+	return cr.shardInfo[groupKey].count
 }
 
 // GetTotalShardCount returns the Shard count of the container.
 func (cr *CachedStore) GetTotalShardCount() int {
 	n := 0
-	for _, v := range cr.resourceInfo {
+	for _, v := range cr.shardInfo {
 		n += v.count
 	}
 	return n
 }
 
-// GetTotalShardCount returns the Shard count of the container.
+// GetGroupKeys returns the Group Key.
 func (cr *CachedStore) GetGroupKeys() string {
 	var v bytes.Buffer
-	for k, vs := range cr.resourceInfo {
+	for k, vs := range cr.shardInfo {
 		v.WriteString(hex.EncodeToString([]byte(k)))
 		v.WriteString(fmt.Sprintf("/%d", vs.count))
 		v.WriteString(" ")
@@ -232,13 +231,13 @@ func (cr *CachedStore) GetTotalLeaderSize() int64 {
 
 // GetShardSize returns the Shard size of the container.
 func (cr *CachedStore) GetShardSize(groupKey string) int64 {
-	return cr.resourceInfo[groupKey].size
+	return cr.shardInfo[groupKey].size
 }
 
 // GetTotalShardSize returns the Shard size of the container.
 func (cr *CachedStore) GetTotalShardSize() int64 {
 	n := int64(0)
-	for _, v := range cr.resourceInfo {
+	for _, v := range cr.shardInfo {
 		n += v.size
 	}
 	return n
@@ -260,7 +259,7 @@ func (cr *CachedStore) GetLeaderWeight() float64 {
 
 // GetShardWeight returns the Shard weight of the container.
 func (cr *CachedStore) GetShardWeight() float64 {
-	return cr.resourceWeight
+	return cr.shardWeight
 }
 
 // GetLastHeartbeatTS returns the last heartbeat timestamp of the container.
@@ -274,7 +273,6 @@ func (cr *CachedStore) NeedPersist() bool {
 }
 
 const minWeight = 1e-6
-const maxScore = 1024 * 1024 * 1024
 
 // LeaderScore returns the container's leader score.
 func (cr *CachedStore) LeaderScore(groupKey string, policy SchedulePolicy, delta int64) float64 {
@@ -292,63 +290,7 @@ func (cr *CachedStore) LeaderScore(groupKey string, policy SchedulePolicy, delta
 // Deviation It is used to control the direction of the deviation considered
 // when calculating the resource score. It is set to -1 when it is the source
 // container of balance, 1 when it is the target, and 0 in the rest of cases.
-func (cr *CachedStore) ShardScore(groupKey string, version string, highSpaceRatio, lowSpaceRatio float64, delta int64, deviation int) float64 {
-	switch version {
-	case "v2":
-		return cr.resourceScoreV2(groupKey, delta, deviation, highSpaceRatio, lowSpaceRatio)
-	case "v1":
-		fallthrough
-	default:
-		return cr.resourceScoreV1(groupKey, highSpaceRatio, lowSpaceRatio, delta)
-	}
-}
-
-func (cr *CachedStore) resourceScoreV1(groupKey string, highSpaceRatio, lowSpaceRatio float64, delta int64) float64 {
-	var score float64
-	var amplification float64
-	available := float64(cr.GetAvailable()) / mb
-	used := float64(cr.GetUsedSize()) / mb
-	capacity := float64(cr.GetCapacity()) / mb
-
-	if cr.GetShardSize(groupKey) == 0 || used == 0 {
-		amplification = 1
-	} else {
-		// because of db compression, resource size is larger than actual used size
-		amplification = float64(cr.GetShardSize(groupKey)) / used
-	}
-
-	// highSpaceBound is the lower bound of the high space stage.
-	highSpaceBound := (1 - highSpaceRatio) * capacity
-	// lowSpaceBound is the upper bound of the low space stage.
-	lowSpaceBound := (1 - lowSpaceRatio) * capacity
-	if available-float64(delta)/amplification >= highSpaceBound {
-		score = float64(cr.GetShardSize(groupKey) + delta)
-	} else if available-float64(delta)/amplification <= lowSpaceBound {
-		score = maxScore - (available - float64(delta)/amplification)
-	} else {
-		// to make the score function continuous, we use linear function y = k * x + b as transition period
-		// from above we know that there are two points must on the function image
-		// note that it is possible that other irrelative files occupy a lot of storage, so capacity == available + used + irrelative
-		// and we regarded irrelative as a fixed value.
-		// Then amp = size / used = size / (capacity - irrelative - available)
-		//
-		// When available == highSpaceBound,
-		// we can conclude that size = (capacity - irrelative - highSpaceBound) * amp = (used + available - highSpaceBound) * amp
-		// Similarly, when available == lowSpaceBound,
-		// we can conclude that size = (capacity - irrelative - lowSpaceBound) * amp = (used + available - lowSpaceBound) * amp
-		// These are the two fixed points' x-coordinates, and y-coordinates which can be easily obtained from the above two functions.
-		x1, y1 := (used+available-highSpaceBound)*amplification, (used+available-highSpaceBound)*amplification
-		x2, y2 := (used+available-lowSpaceBound)*amplification, maxScore-lowSpaceBound
-
-		k := (y2 - y1) / (x2 - x1)
-		b := y1 - k*x1
-		score = k*float64(cr.GetShardSize(groupKey)+delta) + b
-	}
-
-	return score / math.Max(cr.GetShardWeight(), minWeight)
-}
-
-func (cr *CachedStore) resourceScoreV2(groupKey string, delta int64, deviation int, highSpaceRatio, lowSpaceRatio float64) float64 {
+func (cr *CachedStore) ShardScore(groupKey string, highSpaceRatio, lowSpaceRatio float64, delta int64, deviation int) float64 {
 	A := float64(float64(cr.GetAvgAvailable())-float64(deviation)*float64(cr.GetAvailableDeviation())) / gb
 	C := float64(cr.GetCapacity()) / gb
 	R := float64(cr.GetShardSize(groupKey) + delta)
@@ -367,8 +309,8 @@ func (cr *CachedStore) resourceScoreV2(groupKey string, delta int64, deviation i
 		// increases. Ideally, all nodes converge at the position where remaining space is F (default 20GiB).
 		score = (K + M*(math.Log(C)-math.Log(A-F+1))/(C-A+F-1)) * R
 	} else {
-		// When remaining space is less then F, the score is mainly determined by available space.
-		// store's score will increase rapidly after it has few space. and it will reach similar score when they has no space
+		// When remaining space is less than F, the score is mainly determined by available space.
+		// store's score will increase rapidly after it has few space. and it will reach similar score when they have no space
 		score = (K+M*math.Log(C)/C)*R + B*(F-A)/F
 	}
 	return score / math.Max(cr.GetShardWeight(), minWeight)
@@ -443,20 +385,6 @@ func (cr *CachedStore) ShardWeight(kind metapb.ShardType) float64 {
 	}
 }
 
-// GetStartTime returns the start timestamp.
-func (cr *CachedStore) GetStartTime() time.Time {
-	return time.Unix(cr.Meta.GetStartTime(), 0)
-}
-
-// GetUptime returns the uptime.
-func (cr *CachedStore) GetUptime() time.Duration {
-	uptime := cr.GetLastHeartbeatTS().Sub(cr.GetStartTime())
-	if uptime > 0 {
-		return uptime
-	}
-	return 0
-}
-
 var (
 	// If a container's last heartbeat is containerDisconnectDuration ago, the container will
 	// be marked as disconnected state. The value should be greater than storage application's
@@ -487,7 +415,7 @@ func (cr *CachedStore) GetLabelValue(key string) string {
 	return ""
 }
 
-// CompareLocation compares 2 containers' labels and returns at which level their
+// CompareLocation compares 2 stores' labels and returns at which level their
 // locations are different. It returns -1 if they are at the same location.
 func (cr *CachedStore) CompareLocation(other *CachedStore, labels []string) int {
 	for i, key := range labels {
@@ -503,8 +431,8 @@ func (cr *CachedStore) CompareLocation(other *CachedStore, labels []string) int
 
 const replicaBaseScore = 100
 
-// DistinctScore returns the score that the other is distinct from the containers.
-// A higher score means the other container is more different from the existed containers.
+// DistinctScore returns the score that the other is distinct from the stores.
+// A higher score means the other container is more different from the existed stores.
 func DistinctScore(labels []string, containers []*CachedStore, other *CachedStore) float64 {
 	var score float64
 	for _, s := range containers {
@@ -541,139 +469,95 @@ L:
 	return res
 }
 
-// CachedStores contains information about all container.
-type CachedStores struct {
-	containers map[uint64]*CachedStore
+// StoresContainer contains information about all container.
+type StoresContainer struct {
+	stores map[uint64]*CachedStore
 }
 
 // NewCachedStores create a CachedStore with map of containerID to CachedStore
-func NewCachedStores() *CachedStores {
-	return &CachedStores{
-		containers: make(map[uint64]*CachedStore),
+func NewCachedStores() *StoresContainer {
+	return &StoresContainer{
+		stores: make(map[uint64]*CachedStore),
 	}
 }
 
-// GetStore returns a copy of the CachedStore with the specified containerID.
-func (s *CachedStores) GetStore(containerID uint64) *CachedStore {
-	container, ok := s.containers[containerID]
+// GetStore returns CachedStore with the specified containerID.
+func (s *StoresContainer) GetStore(storeID uint64) *CachedStore {
+	store, ok := s.stores[storeID]
 	if !ok {
 		return nil
 	}
-	return container
-}
-
-// TakeStore returns the point of the origin CachedStore with the specified containerID.
-func (s *CachedStores) TakeStore(containerID uint64) *CachedStore {
-	container, ok := s.containers[containerID]
-	if !ok {
-		return nil
-	}
-	return container
+	return store
 }
 
 // SetStore sets a CachedStore with containerID.
-func (s *CachedStores) SetStore(container *CachedStore) {
-	s.containers[container.Meta.GetID()] = container
+func (s *StoresContainer) SetStore(store *CachedStore) {
+	s.stores[store.Meta.GetID()] = store
 }
 
 // PauseLeaderTransfer pauses a CachedStore with containerID.
-func (s *CachedStores) PauseLeaderTransfer(containerID uint64) error {
-	container, ok := s.containers[containerID]
+func (s *StoresContainer) PauseLeaderTransfer(storeID uint64) error {
+	store, ok := s.stores[storeID]
 	if !ok {
-		return fmt.Errorf("container %d not found", containerID)
+		return fmt.Errorf("store %d not found", storeID)
 	}
-	if !container.AllowLeaderTransfer() {
-		return fmt.Errorf("container %d pause transfer leader", containerID)
+	if !store.AllowLeaderTransfer() {
+		return fmt.Errorf("store %d already paused leader transfer", storeID)
 	}
-	s.containers[containerID] = container.Clone(PauseLeaderTransfer())
+	s.stores[storeID] = store.Clone(PauseLeaderTransfer())
 	return nil
 }
 
-// ResumeLeaderTransfer cleans a container's pause state. The container can be selected
+// ResumeLeaderTransfer cleans a store's pause state. The store can be selected
 // as source or target of TransferLeader again.
-func (s *CachedStores) ResumeLeaderTransfer(containerID uint64) {
-	container, ok := s.containers[containerID]
+func (s *StoresContainer) ResumeLeaderTransfer(storeID uint64) {
+	store, ok := s.stores[storeID]
 	if !ok {
-		panic(fmt.Sprintf("try to clean a container %d pause state, but it is not found",
-			containerID))
+		panic(fmt.Sprintf("try to clean a store %d pause state, but it is not found",
+			storeID))
 	}
-	s.containers[containerID] = container.Clone(ResumeLeaderTransfer())
+	s.stores[storeID] = store.Clone(ResumeLeaderTransfer())
 }
 
 // AttachAvailableFunc attaches f to a specific container.
-func (s *CachedStores) AttachAvailableFunc(containerID uint64, limitType limit.Type, f func() bool) {
-	if container, ok := s.containers[containerID]; ok {
-		s.containers[containerID] = container.Clone(AttachAvailableFunc(limitType, f))
+func (s *StoresContainer) AttachAvailableFunc(storeID uint64, limitType limit.Type, f func() bool) {
+	if store, ok := s.stores[storeID]; ok {
+		s.stores[storeID] = store.Clone(AttachAvailableFunc(limitType, f))
 	}
 }
 
 // GetStores gets a complete set of CachedStore.
-func (s *CachedStores) GetStores() []*CachedStore {
-	containers := make([]*CachedStore, 0, len(s.containers))
-	for _, container := range s.containers {
-		containers = append(containers, container)
+func (s *StoresContainer) GetStores() []*CachedStore {
+	stores := make([]*CachedStore, 0, len(s.stores))
+	for _, store := range s.stores {
+		stores = append(stores, store)
 	}
-	return containers
+	return stores
 }
 
 // GetMetaStores gets a complete set of *metapb.Store
-func (s *CachedStores) GetMetaStores() []metapb.Store {
-	metas := make([]metapb.Store, 0, len(s.containers))
-	for _, container := range s.containers {
-		metas = append(metas, container.Meta)
+func (s *StoresContainer) GetMetaStores() []metapb.Store {
+	metas := make([]metapb.Store, 0, len(s.stores))
+	for _, store := range s.stores {
+		metas = append(metas, store.Meta)
 	}
 	return metas
 }
 
-// DeleteStore deletes tombstone record form container
-func (s *CachedStores) DeleteStore(container *CachedStore) {
-	delete(s.containers, container.Meta.GetID())
+// DeleteStore deletes tombstone record form s.stores
+func (s *StoresContainer) DeleteStore(store *CachedStore) {
+	delete(s.stores, store.Meta.GetID())
 }
 
 // GetStoreCount returns the total count of CachedStore.
-func (s *CachedStores) GetStoreCount() int {
-	return len(s.containers)
-}
-
-// SetLeaderCount sets the leader count to a CachedStore.
-func (s *CachedStores) SetLeaderCount(groupKey string, containerID uint64, leaderCount int) {
-	if container, ok := s.containers[containerID]; ok {
-		s.containers[containerID] = container.Clone(SetLeaderCount(groupKey, leaderCount))
-	}
-}
-
-// SetShardCount sets the resource count to a CachedStore.
-func (s *CachedStores) SetShardCount(groupKey string, containerID uint64, resourceCount int) {
-	if container, ok := s.containers[containerID]; ok {
-		s.containers[containerID] = container.Clone(SetShardCount(groupKey, resourceCount))
-	}
-}
-
-// SetPendingPeerCount sets the pending count to a CachedStore.
-func (s *CachedStores) SetPendingPeerCount(groupKey string, containerID uint64, pendingPeerCount int) {
-	if container, ok := s.containers[containerID]; ok {
-		s.containers[containerID] = container.Clone(SetPendingPeerCount(groupKey, pendingPeerCount))
-	}
-}
-
-// SetLeaderSize sets the leader size to a CachedStore.
-func (s *CachedStores) SetLeaderSize(groupKey string, containerID uint64, leaderSize int64) {
-	if container, ok := s.containers[containerID]; ok {
-		s.containers[containerID] = container.Clone(SetLeaderSize(groupKey, leaderSize))
-	}
-}
-
-// SetShardSize sets the resource size to a CachedStore.
-func (s *CachedStores) SetShardSize(groupKey string, containerID uint64, resourceSize int64) {
-	if container, ok := s.containers[containerID]; ok {
-		s.containers[containerID] = container.Clone(SetShardSize(groupKey, resourceSize))
-	}
+func (s *StoresContainer) GetStoreCount() int {
+	return len(s.stores)
 }
 
 // UpdateStoreStatus updates the information of the container.
-func (s *CachedStores) UpdateStoreStatus(groupKey string, containerID uint64, leaderCount int, resourceCount int, pendingPeerCount int, leaderSize int64, resourceSize int64) {
-	if container, ok := s.containers[containerID]; ok {
-		newStore := container.ShallowClone(SetLeaderCount(groupKey, leaderCount),
+func (s *StoresContainer) UpdateStoreStatus(groupKey string, storeID uint64, leaderCount int, resourceCount int, pendingPeerCount int, leaderSize int64, resourceSize int64) {
+	if store, ok := s.stores[storeID]; ok {
+		newStore := store.ShallowClone(SetLeaderCount(groupKey, leaderCount),
 			SetShardCount(groupKey, resourceCount),
 			SetPendingPeerCount(groupKey, pendingPeerCount),
 			SetLeaderSize(groupKey, leaderSize),
diff --git a/components/prophet/core/container_option.go b/components/prophet/core/store_option.go
similarity index 90%
rename from components/prophet/core/container_option.go
rename to components/prophet/core/store_option.go
index 198917d..7ac7e84 100644
--- a/components/prophet/core/container_option.go
+++ b/components/prophet/core/store_option.go
@@ -60,7 +60,7 @@ func SetStoreVersion(commitID, version string) StoreCreateOption {
 	}
 }
 
-// SetStoreDeployPath sets the deploy path for the container.
+// SetStoreDeployPath sets the deploy-path for the container.
 func SetStoreDeployPath(deployPath string) StoreCreateOption {
 	return func(container *CachedStore) {
 		meta := container.Meta.CloneValue()
@@ -97,15 +97,6 @@ func TombstoneStore() StoreCreateOption {
 	}
 }
 
-// SetStoreState sets the state for the container.
-func SetStoreState(state metapb.StoreState) StoreCreateOption {
-	return func(container *CachedStore) {
-		meta := container.Meta.CloneValue()
-		meta.SetState(state)
-		container.Meta = meta
-	}
-}
-
 // PauseLeaderTransfer prevents the container from been selected as source or
 // target container of TransferLeader.
 func PauseLeaderTransfer() StoreCreateOption {
@@ -134,9 +125,9 @@ func SetLeaderCount(groupKey string, leaderCount int) StoreCreateOption {
 // SetShardCount sets the Shard count for the container.
 func SetShardCount(groupKey string, resourceCount int) StoreCreateOption {
 	return func(container *CachedStore) {
-		info := container.resourceInfo[groupKey]
+		info := container.shardInfo[groupKey]
 		info.count = resourceCount
-		container.resourceInfo[groupKey] = info
+		container.shardInfo[groupKey] = info
 	}
 }
 
@@ -159,9 +150,9 @@ func SetLeaderSize(groupKey string, leaderSize int64) StoreCreateOption {
 // SetShardSize sets the Shard size for the container.
 func SetShardSize(groupKey string, resourceSize int64) StoreCreateOption {
 	return func(container *CachedStore) {
-		info := container.resourceInfo[groupKey]
+		info := container.shardInfo[groupKey]
 		info.size = resourceSize
-		container.resourceInfo[groupKey] = info
+		container.shardInfo[groupKey] = info
 	}
 }
 
@@ -175,7 +166,7 @@ func SetLeaderWeight(leaderWeight float64) StoreCreateOption {
 // SetShardWeight sets the Shard weight for the container.
 func SetShardWeight(resourceWeight float64) StoreCreateOption {
 	return func(container *CachedStore) {
-		container.resourceWeight = resourceWeight
+		container.shardWeight = resourceWeight
 	}
 }
 
@@ -196,16 +187,16 @@ func SetLastPersistTime(lastPersist time.Time) StoreCreateOption {
 // SetStoreStats sets the statistics information for the container.
 func SetStoreStats(stats *metapb.StoreStats) StoreCreateOption {
 	return func(container *CachedStore) {
-		container.containerStats.updateRawStats(stats)
+		container.storeStats.updateRawStats(stats)
 	}
 }
 
 // SetNewStoreStats sets the raw statistics information for the container.
 func SetNewStoreStats(stats *metapb.StoreStats) StoreCreateOption {
 	return func(container *CachedStore) {
-		// There is no clone in default container stats, we create new one to avoid to modify others.
+		// There is no clone in default container stats, we create new one to avoid modifying others.
 		// And range cluster cannot use HMA because the last value is not cached
-		container.containerStats = &containerStats{
+		container.storeStats = &storeStats{
 			rawStats: stats,
 		}
 	}
diff --git a/components/prophet/core/container_stats.go b/components/prophet/core/store_stats.go
similarity index 82%
rename from components/prophet/core/container_stats.go
rename to components/prophet/core/store_stats.go
index ebfd3c1..6f87e01 100644
--- a/components/prophet/core/container_stats.go
+++ b/components/prophet/core/store_stats.go
@@ -22,7 +22,7 @@ import (
 	"github.com/matrixorigin/matrixcube/pb/metapb"
 )
 
-type containerStats struct {
+type storeStats struct {
 	mu       sync.RWMutex
 	rawStats *metapb.StoreStats
 
@@ -36,8 +36,8 @@ type containerStats struct {
 	avgMaxAvailableDeviation *movingaverage.HMA
 }
 
-func newStoreStats() *containerStats {
-	return &containerStats{
+func newStoreStats() *storeStats {
+	return &storeStats{
 		rawStats:                 &metapb.StoreStats{},
 		avgAvailable:             movingaverage.NewHMA(240),       // take 40 minutes sample under 10s heartbeat rate
 		maxAvailableDeviation:    movingaverage.NewMaxFilter(120), // take 20 minutes sample under 10s heartbeat rate
@@ -45,7 +45,7 @@ func newStoreStats() *containerStats {
 	}
 }
 
-func (ss *containerStats) updateRawStats(rawStats *metapb.StoreStats) {
+func (ss *storeStats) updateRawStats(rawStats *metapb.StoreStats) {
 	ss.mu.Lock()
 	defer ss.mu.Unlock()
 	ss.rawStats = rawStats
@@ -61,98 +61,98 @@ func (ss *containerStats) updateRawStats(rawStats *metapb.StoreStats) {
 }
 
 // GetStoreStats returns the statistics information of the container.
-func (ss *containerStats) GetStoreStats() *metapb.StoreStats {
+func (ss *storeStats) GetStoreStats() *metapb.StoreStats {
 	ss.mu.RLock()
 	defer ss.mu.RUnlock()
 	return ss.rawStats
 }
 
 // GetCapacity returns the capacity size of the container.
-func (ss *containerStats) GetCapacity() uint64 {
+func (ss *storeStats) GetCapacity() uint64 {
 	ss.mu.RLock()
 	defer ss.mu.RUnlock()
 	return ss.rawStats.GetCapacity()
 }
 
 // GetAvailable returns the available size of the container.
-func (ss *containerStats) GetAvailable() uint64 {
+func (ss *storeStats) GetAvailable() uint64 {
 	ss.mu.RLock()
 	defer ss.mu.RUnlock()
 	return ss.rawStats.GetAvailable()
 }
 
 // GetUsedSize returns the used size of the container.
-func (ss *containerStats) GetUsedSize() uint64 {
+func (ss *storeStats) GetUsedSize() uint64 {
 	ss.mu.RLock()
 	defer ss.mu.RUnlock()
 	return ss.rawStats.GetUsedSize()
 }
 
 // GetUsedRatio returns the used ratio of the container.
-func (ss *containerStats) GetUsedRatio() float64 {
+func (ss *storeStats) GetUsedRatio() float64 {
 	ss.mu.RLock()
 	defer ss.mu.RUnlock()
 	return float64(ss.rawStats.GetUsedSize()) / float64(ss.rawStats.GetCapacity())
 }
 
 // GetBytesWritten returns the bytes written for the container during this period.
-func (ss *containerStats) GetBytesWritten() uint64 {
+func (ss *storeStats) GetBytesWritten() uint64 {
 	ss.mu.RLock()
 	defer ss.mu.RUnlock()
 	return ss.rawStats.GetWrittenBytes()
 }
 
 // GetBytesRead returns the bytes read for the container during this period.
-func (ss *containerStats) GetBytesRead() uint64 {
+func (ss *storeStats) GetBytesRead() uint64 {
 	ss.mu.RLock()
 	defer ss.mu.RUnlock()
 	return ss.rawStats.GetReadBytes()
 }
 
 // GetKeysWritten returns the keys written for the container during this period.
-func (ss *containerStats) GetKeysWritten() uint64 {
+func (ss *storeStats) GetKeysWritten() uint64 {
 	ss.mu.RLock()
 	defer ss.mu.RUnlock()
 	return ss.rawStats.GetWrittenKeys()
 }
 
 // GetKeysRead returns the keys read for the container during this period.
-func (ss *containerStats) GetKeysRead() uint64 {
+func (ss *storeStats) GetKeysRead() uint64 {
 	ss.mu.RLock()
 	defer ss.mu.RUnlock()
 	return ss.rawStats.GetReadKeys()
 }
 
 // IsBusy returns if the container is busy.
-func (ss *containerStats) IsBusy() bool {
+func (ss *storeStats) IsBusy() bool {
 	ss.mu.RLock()
 	defer ss.mu.RUnlock()
 	return ss.rawStats.GetIsBusy()
 }
 
 // GetSendingSnapCount returns the current sending snapshot count of the container.
-func (ss *containerStats) GetSendingSnapCount() uint64 {
+func (ss *storeStats) GetSendingSnapCount() uint64 {
 	ss.mu.RLock()
 	defer ss.mu.RUnlock()
 	return ss.rawStats.GetSendingSnapCount()
 }
 
 // GetReceivingSnapCount returns the current receiving snapshot count of the container.
-func (ss *containerStats) GetReceivingSnapCount() uint64 {
+func (ss *storeStats) GetReceivingSnapCount() uint64 {
 	ss.mu.RLock()
 	defer ss.mu.RUnlock()
 	return ss.rawStats.GetReceivingSnapCount()
 }
 
 // GetApplyingSnapCount returns the current applying snapshot count of the container.
-func (ss *containerStats) GetApplyingSnapCount() uint64 {
+func (ss *storeStats) GetApplyingSnapCount() uint64 {
 	ss.mu.RLock()
 	defer ss.mu.RUnlock()
 	return ss.rawStats.GetApplyingSnapCount()
 }
 
 // GetAvgAvailable returns available size after the spike changes has been smoothed.
-func (ss *containerStats) GetAvgAvailable() uint64 {
+func (ss *storeStats) GetAvgAvailable() uint64 {
 	ss.mu.RLock()
 	defer ss.mu.RUnlock()
 	if ss.avgAvailable == nil {
@@ -162,7 +162,7 @@ func (ss *containerStats) GetAvgAvailable() uint64 {
 }
 
 // GetAvailableDeviation returns approximate magnitude of available in the recent period.
-func (ss *containerStats) GetAvailableDeviation() uint64 {
+func (ss *storeStats) GetAvailableDeviation() uint64 {
 	ss.mu.RLock()
 	defer ss.mu.RUnlock()
 	if ss.avgMaxAvailableDeviation == nil {
diff --git a/components/prophet/mock/mockjob/job.go b/components/prophet/mock/mockjob/job.go
index d160c70..2078809 100644
--- a/components/prophet/mock/mockjob/job.go
+++ b/components/prophet/mock/mockjob/job.go
@@ -49,16 +49,16 @@ func (mr *MockShardsAwareMockRecorder) ForeachShards(group, fn interface{}) *gom
 	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "ForeachShards", reflect.TypeOf((*MockShardsAware)(nil).ForeachShards), group, fn)
 }
 
-// ForeachWaittingCreateShards mocks base method.
-func (m *MockShardsAware) ForeachWaittingCreateShards(do func(metapb.Shard)) {
+// ForeachWaitingCreateShards mocks base method.
+func (m *MockShardsAware) ForeachWaitingCreateShards(do func(metapb.Shard)) {
 	m.ctrl.T.Helper()
-	m.ctrl.Call(m, "ForeachWaittingCreateShards", do)
+	m.ctrl.Call(m, "ForeachWaitingCreateShards", do)
 }
 
-// ForeachWaittingCreateShards indicates an expected call of ForeachWaittingCreateShards.
-func (mr *MockShardsAwareMockRecorder) ForeachWaittingCreateShards(do interface{}) *gomock.Call {
+// ForeachWaitingCreateShards indicates an expected call of ForeachWaitingCreateShards.
+func (mr *MockShardsAwareMockRecorder) ForeachWaitingCreateShards(do interface{}) *gomock.Call {
 	mr.mock.ctrl.T.Helper()
-	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "ForeachWaittingCreateShards", reflect.TypeOf((*MockShardsAware)(nil).ForeachWaittingCreateShards), do)
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "ForeachWaitingCreateShards", reflect.TypeOf((*MockShardsAware)(nil).ForeachWaitingCreateShards), do)
 }
 
 // GetShard mocks base method.
diff --git a/components/prophet/schedule/filter/comparer.go b/components/prophet/schedule/filter/comparer.go
index 4131437..4ea8381 100644
--- a/components/prophet/schedule/filter/comparer.go
+++ b/components/prophet/schedule/filter/comparer.go
@@ -27,8 +27,8 @@ type StoreComparer func(a, b *core.CachedStore) int
 // score.
 func ShardScoreComparer(groupKey string, opt *config.PersistOptions) StoreComparer {
 	return func(a, b *core.CachedStore) int {
-		sa := a.ShardScore(groupKey, opt.GetShardScoreFormulaVersion(), opt.GetHighSpaceRatio(), opt.GetLowSpaceRatio(), 0, 0)
-		sb := b.ShardScore(groupKey, opt.GetShardScoreFormulaVersion(), opt.GetHighSpaceRatio(), opt.GetLowSpaceRatio(), 0, 0)
+		sa := a.ShardScore(groupKey, opt.GetHighSpaceRatio(), opt.GetLowSpaceRatio(), 0, 0)
+		sb := b.ShardScore(groupKey, opt.GetHighSpaceRatio(), opt.GetLowSpaceRatio(), 0, 0)
 		switch {
 		case sa > sb:
 			return 1
diff --git a/components/prophet/schedule/opt/healthy.go b/components/prophet/schedule/opt/healthy.go
index 67489b8..57a01eb 100644
--- a/components/prophet/schedule/opt/healthy.go
+++ b/components/prophet/schedule/opt/healthy.go
@@ -40,7 +40,7 @@ func IsHealthyAllowPending(cluster Cluster, res *core.CachedShard) bool {
 
 // IsEmptyShardAllowBalance checks if a region is an empty region and can be balanced.
 func IsEmptyShardAllowBalance(cluster Cluster, res *core.CachedShard) bool {
-	return res.GetApproximateSize() > core.EmptyShardApproximateSize ||
+	return res.GetApproximateSize() > core.MinShardSize ||
 		cluster.GetShardCount() < balanceEmptyShardThreshold
 }
 
diff --git a/components/prophet/schedulers/balance_resource.go b/components/prophet/schedulers/balance_resource.go
index 1a36e64..17edadf 100644
--- a/components/prophet/schedulers/balance_resource.go
+++ b/components/prophet/schedulers/balance_resource.go
@@ -167,8 +167,8 @@ func (s *balanceShardScheduler) scheduleByGroup(groupKey string, cluster opt.Clu
 	sort.Slice(containers, func(i, j int) bool {
 		iOp := opInfluence.GetStoreInfluence(containers[i].Meta.GetID()).ShardProperty(kind, groupKey)
 		jOp := opInfluence.GetStoreInfluence(containers[j].Meta.GetID()).ShardProperty(kind, groupKey)
-		return containers[i].ShardScore(groupKey, opts.GetShardScoreFormulaVersion(), opts.GetHighSpaceRatio(), opts.GetLowSpaceRatio(), iOp, -1) >
-			containers[j].ShardScore(groupKey, opts.GetShardScoreFormulaVersion(), opts.GetHighSpaceRatio(), opts.GetLowSpaceRatio(), jOp, -1)
+		return containers[i].ShardScore(groupKey, opts.GetHighSpaceRatio(), opts.GetLowSpaceRatio(), iOp, -1) >
+			containers[j].ShardScore(groupKey, opts.GetHighSpaceRatio(), opts.GetLowSpaceRatio(), jOp, -1)
 	})
 
 	groupID := util.DecodeGroupKey(groupKey)
diff --git a/components/prophet/schedulers/utils.go b/components/prophet/schedulers/utils.go
index 08bee3a..39f6a48 100644
--- a/components/prophet/schedulers/utils.go
+++ b/components/prophet/schedulers/utils.go
@@ -69,8 +69,8 @@ func shouldBalance(cluster opt.Cluster,
 		sourceScore = source.LeaderScore(res.GetGroupKey(), kind.Policy, sourceDelta)
 		targetScore = target.LeaderScore(res.GetGroupKey(), kind.Policy, targetDelta)
 	case metapb.ShardType_AllShards:
-		sourceScore = source.ShardScore(res.GetGroupKey(), opts.GetShardScoreFormulaVersion(), opts.GetHighSpaceRatio(), opts.GetLowSpaceRatio(), sourceDelta, -1)
-		targetScore = target.ShardScore(res.GetGroupKey(), opts.GetShardScoreFormulaVersion(), opts.GetHighSpaceRatio(), opts.GetLowSpaceRatio(), targetDelta, 1)
+		sourceScore = source.ShardScore(res.GetGroupKey(), opts.GetHighSpaceRatio(), opts.GetLowSpaceRatio(), sourceDelta, -1)
+		targetScore = target.ShardScore(res.GetGroupKey(), opts.GetHighSpaceRatio(), opts.GetLowSpaceRatio(), targetDelta, 1)
 	}
 	if opts.IsDebugMetricsEnabled() {
 		opInfluenceStatus.WithLabelValues(scheduleName, strconv.FormatUint(sourceID, 10), "source").Set(float64(sourceInfluence))
@@ -155,7 +155,7 @@ func getKeyRanges(args []string) ([]core.KeyRange, error) {
 	for len(args) > 1 {
 		groupID, err := strconv.ParseUint(args[0], 10, 64)
 		if err != nil {
-			return nil, errors.New("scheduler error coniguration")
+			return nil, errors.New("scheduler error configuration")
 		}
 
 		startKey, err := url.QueryUnescape(args[1])
diff --git a/components/prophet/statistics/resource.go b/components/prophet/statistics/resource.go
index c678107..ee58fe3 100644
--- a/components/prophet/statistics/resource.go
+++ b/components/prophet/statistics/resource.go
@@ -57,7 +57,7 @@ func (s *ShardStats) Observe(r *core.CachedShard) {
 	s.Count++
 	approximateKeys := r.GetApproximateKeys()
 	approximateSize := r.GetApproximateSize()
-	if approximateSize <= core.EmptyShardApproximateSize {
+	if approximateSize <= core.MinShardSize {
 		s.EmptyCount++
 	}
 	s.StorageSize += approximateSize
diff --git a/components/prophet/statistics/resource_collection.go b/components/prophet/statistics/resource_collection.go
index ae8939c..f578820 100644
--- a/components/prophet/statistics/resource_collection.go
+++ b/components/prophet/statistics/resource_collection.go
@@ -147,7 +147,7 @@ func (r *ShardStatistics) Observe(res *core.CachedShard, containers []*core.Cach
 		DownPeer:    len(res.GetDownPeers()) > 0,
 		PendingPeer: len(res.GetPendingPeers()) > 0,
 		LearnerPeer: len(res.GetLearners()) > 0,
-		EmptyShard:  res.GetApproximateSize() <= core.EmptyShardApproximateSize,
+		EmptyShard:  res.GetApproximateSize() <= core.MinShardSize,
 	}
 
 	for typ, c := range conditions {
diff --git a/components/prophet/storage/kv.go b/components/prophet/storage/kv.go
index 5d0607e..c0fd888 100644
--- a/components/prophet/storage/kv.go
+++ b/components/prophet/storage/kv.go
@@ -36,16 +36,8 @@ type KV interface {
 	Remove(key string) error
 	// LoadRange iterates all key-value pairs in the storage
 	LoadRange(key, endKey string, limit int64) ([]string, []string, error)
-	// CountRange count all key-value pairs in the storage
-	CountRange(key, endKey string) (uint64, error)
 	// SaveIfNotExists put the value at path
 	// returns true, nil, nil if created
 	// returns false, exists, nil if not created
 	SaveIfNotExists(key string, value string, batch *Batch) (bool, string, error)
-	// RemoveIfValueMatched returns true if the expect value is and the exists value are matched
-	RemoveIfValueMatched(key string, expect string) (bool, error)
-	// SaveWithoutLeader save without leader
-	SaveWithoutLeader(key, value string) error
-	// RemoveWithoutLeader remove without leader
-	RemoveWithoutLeader(key string) error
 }
diff --git a/components/prophet/storage/kv_etcd.go b/components/prophet/storage/kv_etcd.go
index c289a67..8a9c7be 100644
--- a/components/prophet/storage/kv_etcd.go
+++ b/components/prophet/storage/kv_etcd.go
@@ -126,23 +126,6 @@ func (s *etcdKV) LoadRange(key, endKey string, limit int64) ([]string, []string,
 	return keys, values, nil
 }
 
-func (s *etcdKV) CountRange(key, endKey string) (uint64, error) {
-	// Note: reason to use `strings.Join` instead of `path.Join` is that the latter will
-	// removes suffix '/' of the joined string.
-	// As a result, when we try to scan from "foo/", it ends up scanning from "/pd/foo"
-	// internally, and returns unexpected keys such as "foo_bar/baz".
-	key = strings.Join([]string{key}, "/")
-	endKey = strings.Join([]string{endKey}, "/")
-
-	withRange := clientv3.WithRange(endKey)
-	resp, err := util.GetEtcdResp(s.client, key, withRange, clientv3.WithCountOnly())
-	if err != nil {
-		return 0, err
-	}
-
-	return uint64(resp.Count), nil
-}
-
 func (s *etcdKV) SaveIfNotExists(key string, value string, batch *Batch) (bool, string, error) {
 	var ops []clientv3.Op
 	ops = append(ops, clientv3.OpPut(key, value))
@@ -175,43 +158,3 @@ func (s *etcdKV) SaveIfNotExists(key string, value string, batch *Batch) (bool,
 
 	return true, "", nil
 }
-
-func (s *etcdKV) RemoveIfValueMatched(key string, expect string) (bool, error) {
-	resp, err := util.Txn(s.client).
-		If(clientv3.Compare(clientv3.Value(key), "=", string(expect))).
-		Then(clientv3.OpDelete(key)).
-		Commit()
-	if err != nil {
-		return false, err
-	}
-
-	if !resp.Succeeded {
-		return false, nil
-	}
-
-	return true, nil
-}
-
-func (s *etcdKV) SaveWithoutLeader(key, value string) error {
-	var ops []clientv3.Op
-	ops = append(ops, clientv3.OpPut(key, value))
-
-	_, err := util.Txn(s.client).Then(ops...).Commit()
-	if err != nil {
-		return err
-	}
-
-	return nil
-}
-
-func (s *etcdKV) RemoveWithoutLeader(key string) error {
-	var ops []clientv3.Op
-	ops = append(ops, clientv3.OpDelete(key))
-
-	_, err := util.Txn(s.client).Then(ops...).Commit()
-	if err != nil {
-		return err
-	}
-
-	return nil
-}
diff --git a/components/prophet/storage/kv_mem.go b/components/prophet/storage/kv_mem.go
index 3aa1a5a..7e30198 100644
--- a/components/prophet/storage/kv_mem.go
+++ b/components/prophet/storage/kv_mem.go
@@ -93,19 +93,6 @@ func (s *memStorage) LoadRange(key, endKey string, limit int64) ([]string, []str
 	return keys, values, nil
 }
 
-func (s *memStorage) CountRange(key, endKey string) (uint64, error) {
-	s.RLock()
-	defer s.RUnlock()
-
-	cnt := uint64(0)
-	s.tree.AscendRange(memoryKVItem{key, ""}, memoryKVItem{endKey, ""}, func(item btree.Item) bool {
-		cnt++
-		return true
-	})
-
-	return cnt, nil
-}
-
 func (s *memStorage) SaveIfNotExists(key string, value string, batch *Batch) (bool, string, error) {
 	s.Lock()
 	defer s.Unlock()
@@ -137,41 +124,6 @@ func (s *memStorage) SaveIfNotExists(key string, value string, batch *Batch) (bo
 	return true, "", err
 }
 
-func (s *memStorage) RemoveIfValueMatched(key string, expect string) (bool, error) {
-	s.Lock()
-	defer s.Unlock()
-
-	old, err := s.doLoad(key)
-	if err != nil {
-		return false, err
-	}
-
-	if old != expect {
-		return false, nil
-	}
-
-	err = s.doRemove(key)
-	if err != nil {
-		return false, err
-	}
-
-	return true, nil
-}
-
-func (s *memStorage) SaveWithoutLeader(key, value string) error {
-	s.Lock()
-	defer s.Unlock()
-
-	return s.doSave(string(key), string(value))
-}
-
-func (s *memStorage) RemoveWithoutLeader(key string) error {
-	s.Lock()
-	defer s.Unlock()
-
-	return s.doRemove(key)
-}
-
 func (s *memStorage) doLoad(key string) (string, error) {
 	item := s.tree.Get(memoryKVItem{key, ""})
 	if item == nil {
diff --git a/components/prophet/storage/storage.go b/components/prophet/storage/storage.go
index d7336c0..bb492ca 100644
--- a/components/prophet/storage/storage.go
+++ b/components/prophet/storage/storage.go
@@ -63,18 +63,6 @@ type RuleStorage interface {
 	LoadRuleGroups(limit int64, f func(k, v string) error) error
 }
 
-// CustomDataStorage custom data storage
-type CustomDataStorage interface {
-	// PutCustomData puts the custom data to the storage
-	PutCustomData(key []byte, data []byte) error
-	// BatchPutCustomData batch puts the custom data to the storage
-	BatchPutCustomData(keys [][]byte, data [][]byte) error
-	// LoadCustomData load all custom data
-	LoadCustomData(limit int64, f func(k, v []byte) error) error
-	// RemoveCustomData remove custom data
-	RemoveCustomData(key []byte) error
-}
-
 // ShardStorage resource storage
 type ShardStorage interface {
 	// PutShard puts the meta to the storage
@@ -141,7 +129,6 @@ type ClusterStorage interface {
 // Storage meta storage
 type Storage interface {
 	JobStorage
-	CustomDataStorage
 	RuleStorage
 	ConfigStorage
 	StoreStorage
@@ -172,7 +159,6 @@ type storage struct {
 	schedulePath             string
 	jobPath                  string
 	jobDataPath              string
-	customDataPath           string
 }
 
 // NewTestStorage create test storage
@@ -198,7 +184,6 @@ func NewStorage(rootPath string, kv KV, idGen id.Generator) Storage {
 		schedulePath:             fmt.Sprintf("%s/schedule", rootPath),
 		jobPath:                  fmt.Sprintf("%s/jobs", rootPath),
 		jobDataPath:              fmt.Sprintf("%s/job-data", rootPath),
-		customDataPath:           fmt.Sprintf("%s/custom", rootPath),
 	}
 }
 
@@ -513,36 +498,6 @@ func (s *storage) RemoveJobData(job metapb.Job) error {
 	return s.kv.Remove(s.jobDataKey(job.Type))
 }
 
-func (s *storage) PutCustomData(key []byte, data []byte) error {
-	return s.kv.Save(path.Join(s.customDataPath, string(key)), string(data))
-}
-
-func (s *storage) BatchPutCustomData(keys [][]byte, data [][]byte) error {
-	if len(keys) != len(data) {
-		return fmt.Errorf("key length %d != data length %d",
-			len(keys),
-			len(data))
-	}
-
-	batch := &Batch{}
-	for i := 0; i < len(keys); i++ {
-		batch.SaveKeys = append(batch.SaveKeys, path.Join(s.customDataPath, string(keys[i])))
-		batch.SaveValues = append(batch.SaveValues, string(data[i]))
-	}
-	return s.kv.Batch(batch)
-}
-
-func (s *storage) LoadCustomData(limit int64, do func(k, v []byte) error) error {
-	return s.LoadRangeByPrefix(limit, s.customDataPath+"/", func(k, v string) error {
-		do([]byte(k), []byte(v))
-		return nil
-	})
-}
-
-func (s *storage) RemoveCustomData(key []byte) error {
-	return s.kv.Remove(path.Join(s.customDataPath, string(key)))
-}
-
 func (s *storage) PutBootstrapped(container metapb.Store, resources ...*metapb.Shard) (bool, error) {
 	clusterID, err := s.idGen.AllocID()
 	if err != nil {
diff --git a/raftstore/store_shards_pool.go b/raftstore/store_shards_pool.go
index 45c72c8..80ec945 100644
--- a/raftstore/store_shards_pool.go
+++ b/raftstore/store_shards_pool.go
@@ -234,7 +234,7 @@ func (dsp *dynamicShardsPool) doAllocLocked(cmd *metapb.ShardsPoolAllocCmd, stor
 			id = shard.ID
 		}
 	}
-	aware.ForeachWaittingCreateShards(fn)
+	aware.ForeachWaitingCreateShards(fn)
 	if id == 0 {
 		aware.ForeachShards(group, fn)
 	}


Test patch:
diff --git a/components/prophet/cluster/cluster_test.go b/components/prophet/cluster/cluster_test.go
index 4613f8e..99daae3 100644
--- a/components/prophet/cluster/cluster_test.go
+++ b/components/prophet/cluster/cluster_test.go
@@ -381,7 +381,7 @@ func TestShardHeartbeat(t *testing.T) {
 		checkShard(t, res, resources[res.Meta.GetID()])
 	}
 	for _, res := range cluster.GetMetaShards() {
-		assert.True(t, reflect.DeepEqual(*res, resources[res.GetID()].Meta))
+		assert.True(t, reflect.DeepEqual(res, resources[res.GetID()].Meta))
 	}
 
 	for _, res := range resources {
@@ -889,7 +889,7 @@ func checkShardsKV(t *testing.T, s storage.Storage, resources []*core.CachedShar
 	}
 }
 
-func checkShards(t *testing.T, cache *core.CachedShards, resources []*core.CachedShard) {
+func checkShards(t *testing.T, cache *core.ShardsContainer, resources []*core.CachedShard) {
 	resourceCount := make(map[uint64]int)
 	leaderCount := make(map[uint64]int)
 	followerCount := make(map[uint64]int)
@@ -921,7 +921,7 @@ func checkShards(t *testing.T, cache *core.CachedShards, resources []*core.Cache
 		checkShard(t, res, resources[res.Meta.GetID()])
 	}
 	for _, res := range cache.GetMetaShards() {
-		assert.True(t, reflect.DeepEqual(*res, resources[res.GetID()].Meta))
+		assert.True(t, reflect.DeepEqual(res, resources[res.GetID()].Meta))
 	}
 }
 
diff --git a/components/prophet/cluster/cluster_worker_test.go b/components/prophet/cluster/cluster_worker_test.go
index ec246e2..0db87aa 100644
--- a/components/prophet/cluster/cluster_worker_test.go
+++ b/components/prophet/cluster/cluster_worker_test.go
@@ -65,13 +65,13 @@ func TestCreateShards(t *testing.T) {
 	cluster.addShardStore(3, 1)
 	_, err = cluster.HandleCreateShards(req)
 	assert.NoError(t, err)
-	assert.Equal(t, 1, len(cluster.core.WaittingCreateShards))
+	assert.Equal(t, 1, len(cluster.core.WaitingCreateShards))
 	assert.Equal(t, 0, cluster.GetShardCount())
 
 	// recreate
 	_, err = cluster.HandleCreateShards(req)
 	assert.NoError(t, err)
-	assert.Equal(t, 1, len(cluster.core.WaittingCreateShards))
+	assert.Equal(t, 1, len(cluster.core.WaitingCreateShards))
 	assert.Equal(t, 0, cluster.GetShardCount())
 
 	cluster.doNotifyCreateShards()
@@ -79,14 +79,14 @@ func TestCreateShards(t *testing.T) {
 	assert.Equal(t, event.EventShard, e.Type)
 	assert.True(t, e.ShardEvent.Create)
 
-	for _, res := range cluster.core.WaittingCreateShards {
+	for _, res := range cluster.core.WaitingCreateShards {
 		v, err := cluster.storage.GetShard(res.GetID())
 		assert.NoError(t, err)
 		assert.Equal(t, metapb.ShardState_Creating, v.GetState())
 
 		assert.NoError(t, cluster.HandleShardHeartbeat(core.NewCachedShard(res, &res.GetReplicas()[0])))
 		assert.Equal(t, 1, cluster.GetShardCount())
-		assert.Equal(t, 0, len(cluster.core.WaittingCreateShards))
+		assert.Equal(t, 0, len(cluster.core.WaitingCreateShards))
 		assert.NotNil(t, changedRes)
 		assert.Equal(t, changedResFrom, metapb.ShardState_Creating)
 		assert.Equal(t, changedResTo, metapb.ShardState_Running)
@@ -118,7 +118,7 @@ func TestCreateShardsRestart(t *testing.T) {
 	// restart
 	tc := newTestRaftCluster(cluster.GetOpts(), cluster.storage, core.NewBasicCluster(nil))
 	tc.LoadClusterInfo()
-	assert.Equal(t, 1, len(tc.core.WaittingCreateShards))
+	assert.Equal(t, 1, len(tc.core.WaitingCreateShards))
 	tc.doNotifyCreateShards()
 	e := <-tc.ChangedEventNotifier()
 	assert.Equal(t, event.EventShard, e.Type)
diff --git a/components/prophet/core/resource_test.go b/components/prophet/core/shard_test.go
similarity index 97%
rename from components/prophet/core/resource_test.go
rename to components/prophet/core/shard_test.go
index e4efe3a..15519b6 100644
--- a/components/prophet/core/resource_test.go
+++ b/components/prophet/core/shard_test.go
@@ -109,7 +109,7 @@ func TestSortedEqual(t *testing.T) {
 }
 
 func TestShardMap(t *testing.T) {
-	var empty *resourceMap
+	var empty *shardMap
 	assert.Equal(t, 0, empty.Len(), "TestShardMap failed")
 	assert.Nil(t, empty.Get(1), "TestShardMap failed")
 
@@ -236,8 +236,8 @@ func TestSetShard(t *testing.T) {
 	assert.Equal(t, 96, resources.trees[0].length(), "TestSetShard failed")
 	assert.Equal(t, 96, len(resources.GetShards()), "TestSetShard failed")
 	assert.NotNil(t, resources.GetShard(201), "TestSetShard failed")
-	assert.Equal(t, int64(20), resources.resources.totalKeys, "TestSetShard failed")
-	assert.Equal(t, int64(30), resources.resources.totalSize, "TestSetShard failed")
+	assert.Equal(t, int64(20), resources.shards.totalKeys, "TestSetShard failed")
+	assert.Equal(t, int64(30), resources.shards.totalSize, "TestSetShard failed")
 }
 
 func TestShouldRemoveFromSubTree(t *testing.T) {
@@ -280,7 +280,7 @@ func TestShouldRemoveFromSubTree(t *testing.T) {
 	assert.True(t, resources.shouldRemoveFromSubTree(res, origin))
 }
 
-func checkShardMap(t *testing.T, msg string, rm *resourceMap, ids ...uint64) {
+func checkShardMap(t *testing.T, msg string, rm *shardMap, ids ...uint64) {
 	// Check Get.
 	for _, id := range ids {
 		assert.Equal(t, id, rm.Get(id).Meta.GetID(), msg)
@@ -307,7 +307,7 @@ func checkShardMap(t *testing.T, msg string, rm *resourceMap, ids ...uint64) {
 	assert.Equal(t, rm.totalSize, total, msg)
 }
 
-func checkShards(t *testing.T, resources *CachedShards, msg string) {
+func checkShards(t *testing.T, resources *ShardsContainer, msg string) {
 	leaderMap := make(map[uint64]uint64)
 	followerMap := make(map[uint64]uint64)
 	learnerMap := make(map[uint64]uint64)
diff --git a/components/prophet/core/resource_tree_test.go b/components/prophet/core/shard_tree_test.go
similarity index 94%
rename from components/prophet/core/resource_tree_test.go
rename to components/prophet/core/shard_tree_test.go
index 3529647..7b6ad8b 100644
--- a/components/prophet/core/resource_tree_test.go
+++ b/components/prophet/core/shard_tree_test.go
@@ -202,35 +202,35 @@ func TestShardTree(t *testing.T) {
 	assert.Nil(t, tree.search([]byte("c")))
 	assert.Equal(t, resD, tree.search([]byte("d")))
 
-	// check get adjacent resources
+	// check get adjacent shards
 	prev, next := tree.getAdjacentShards(resA)
 	assert.Nil(t, prev)
-	assert.Equal(t, resB, next.res)
+	assert.Equal(t, resB, next.shard)
 
 	prev, next = tree.getAdjacentShards(resB)
-	assert.Equal(t, resA, prev.res)
-	assert.Equal(t, resD, next.res)
+	assert.Equal(t, resA, prev.shard)
+	assert.Equal(t, resD, next.shard)
 
 	prev, next = tree.getAdjacentShards(resC)
-	assert.Equal(t, resB, prev.res)
-	assert.Equal(t, resD, next.res)
+	assert.Equal(t, resB, prev.shard)
+	assert.Equal(t, resD, next.shard)
 
 	prev, next = tree.getAdjacentShards(resD)
-	assert.Equal(t, resB, prev.res)
+	assert.Equal(t, resB, prev.shard)
 	assert.Nil(t, next)
 
-	// resource with the same range and different resource id will not be delete.
-	res0 := newTestShardItem([]byte{}, []byte("a")).res
+	// resource with the same range and different resource id will not be deleted.
+	res0 := newTestShardItem([]byte{}, []byte("a")).shard
 	tree.update(res0)
 	assert.Equal(t, res0, tree.search([]byte{}))
 
-	anotherRes0 := newTestShardItem([]byte{}, []byte("a")).res
+	anotherRes0 := newTestShardItem([]byte{}, []byte("a")).shard
 	anotherRes0.Meta.SetID(123)
 	tree.remove(anotherRes0)
 	assert.Equal(t, res0, tree.search([]byte{}))
 
 	// overlaps with 0, A, B, C.
-	res0D := newTestShardItem([]byte(""), []byte("d")).res
+	res0D := newTestShardItem([]byte(""), []byte("d")).shard
 	tree.update(res0D)
 	assert.Equal(t, res0D, tree.search([]byte{}))
 	assert.Equal(t, res0D, tree.search([]byte("a")))
@@ -239,7 +239,7 @@ func TestShardTree(t *testing.T) {
 	assert.Equal(t, resD, tree.search([]byte("d")))
 
 	// overlaps with D.
-	resE := newTestShardItem([]byte("e"), []byte{}).res
+	resE := newTestShardItem([]byte("e"), []byte{}).shard
 	tree.update(resE)
 	assert.Equal(t, res0D, tree.search([]byte{}))
 	assert.Equal(t, res0D, tree.search([]byte("a")))
@@ -251,7 +251,7 @@ func TestShardTree(t *testing.T) {
 
 func TestShardTreeSplitAndMerge(t *testing.T) {
 	tree := newShardTree()
-	resources := []*CachedShard{newTestShardItem([]byte{}, []byte{}).res}
+	resources := []*CachedShard{newTestShardItem([]byte{}, []byte{}).shard}
 
 	// Byte will underflow/overflow if n > 7.
 	n := 7
@@ -376,7 +376,7 @@ func TestRandomShardDiscontinuous(t *testing.T) {
 	checkRandomShard(t, tree, []*CachedShard{resA, resB, resC, resD}, []KeyRange{NewKeyRange(0, "", "")})
 }
 
-func updateTestShards(t *testing.T, tree *resourceTree, resources []*CachedShard) {
+func updateTestShards(t *testing.T, tree *shardTree, resources []*CachedShard) {
 	for _, res := range resources {
 		startKey, endKey := res.Meta.GetRange()
 		tree.update(res)
@@ -389,7 +389,7 @@ func updateTestShards(t *testing.T, tree *resourceTree, resources []*CachedShard
 	}
 }
 
-func checkRandomShard(t *testing.T, tree *resourceTree, resources []*CachedShard, ranges []KeyRange) {
+func checkRandomShard(t *testing.T, tree *shardTree, resources []*CachedShard, ranges []KeyRange) {
 	keys := make(map[string]struct{})
 	for i := 0; i < 10000 && len(keys) < len(resources); i++ {
 		re := tree.RandomShard(ranges)
diff --git a/components/prophet/core/container_stats_test.go b/components/prophet/core/store_stats_test.go
similarity index 100%
rename from components/prophet/core/container_stats_test.go
rename to components/prophet/core/store_stats_test.go
diff --git a/components/prophet/core/container_test.go b/components/prophet/core/store_test.go
similarity index 90%
rename from components/prophet/core/container_test.go
rename to components/prophet/core/store_test.go
index ec079c8..c959af1 100644
--- a/components/prophet/core/container_test.go
+++ b/components/prophet/core/store_test.go
@@ -43,11 +43,11 @@ func TestDistinctScore(t *testing.T) {
 				container := NewTestStoreInfoWithLabel(containerID, 1, containerLabels)
 				containers = append(containers, container)
 
-				// Number of containers in different zones.
+				// Number of stores in different zones.
 				numZones := i * len(racks) * len(hosts)
-				// Number of containers in the same zone but in different racks.
+				// Number of stores in the same zone but in different racks.
 				numRacks := j * len(hosts)
-				// Number of containers in the same rack but in different hosts.
+				// Number of stores in the same rack but in different hosts.
 				numHosts := k
 				score := (numZones*replicaBaseScore+numRacks)*replicaBaseScore + numHosts
 				assert.Equal(t, float64(score), DistinctScore(labels, containers, container))
@@ -101,7 +101,7 @@ func TestShardScore(t *testing.T) {
 		SetStoreStats(stats),
 		SetShardSize("", 1),
 	)
-	score := container.ShardScore("", "v1", 0.7, 0.9, 0, 0)
+	score := container.ShardScore("", 0.7, 0.9, 0, 0)
 	// Shard score should never be NaN, or /container API would fail.
 	assert.False(t, math.IsNaN(score))
 }
@@ -112,9 +112,9 @@ func TestLowSpaceRatio(t *testing.T) {
 	container.rawStats.Available = container.rawStats.Capacity >> 3
 
 	assert.False(t, container.IsLowSpace(0.8))
-	info := container.resourceInfo[""]
+	info := container.shardInfo[""]
 	info.count = 31
-	container.resourceInfo[""] = info
+	container.shardInfo[""] = info
 	assert.True(t, container.IsLowSpace(0.8))
 	container.rawStats.Available = container.rawStats.Capacity >> 2
 	assert.False(t, container.IsLowSpace(0.8))
diff --git a/components/prophet/core/test_util.go b/components/prophet/core/test_util.go
index 3c6ec73..ce7c0cc 100644
--- a/components/prophet/core/test_util.go
+++ b/components/prophet/core/test_util.go
@@ -20,10 +20,10 @@ import (
 	"github.com/matrixorigin/matrixcube/pb/metapb"
 )
 
-// SplitTestShards split a set of CachedShard by the middle of resourceKey
-func SplitTestShards(resources []*CachedShard) []*CachedShard {
-	results := make([]*CachedShard, 0, len(resources)*2)
-	for _, res := range resources {
+// SplitTestShards split a set of CachedShard by the middle of shardKey
+func SplitTestShards(shards []*CachedShard) []*CachedShard {
+	results := make([]*CachedShard, 0, len(shards)*2)
+	for _, res := range shards {
 		resStart, resEnd := res.Meta.GetRange()
 		start, end := byte(0), byte(math.MaxUint8)
 		if len(resStart) > 0 {
@@ -35,14 +35,14 @@ func SplitTestShards(resources []*CachedShard) []*CachedShard {
 		middle := []byte{start/2 + end/2}
 
 		left := res.Clone()
-		left.Meta.SetID(res.Meta.GetID() + uint64(len(resources)))
+		left.Meta.SetID(res.Meta.GetID() + uint64(len(shards)))
 		left.Meta.SetEndKey(middle)
 		epoch := left.Meta.GetEpoch()
 		epoch.Generation++
 		left.Meta.SetEpoch(epoch)
 
 		right := res.Clone()
-		right.Meta.SetID(res.Meta.GetID() + uint64(len(resources)*2))
+		right.Meta.SetID(res.Meta.GetID() + uint64(len(shards)*2))
 		right.Meta.SetStartKey(middle)
 		epoch = right.Meta.GetEpoch()
 		epoch.Generation++
@@ -52,21 +52,21 @@ func SplitTestShards(resources []*CachedShard) []*CachedShard {
 	return results
 }
 
-// MergeTestShards merge a set of CachedShard by resourceKey
-func MergeTestShards(resources []*CachedShard) []*CachedShard {
-	results := make([]*CachedShard, 0, len(resources)/2)
-	for i := 0; i < len(resources); i += 2 {
-		left := resources[i]
-		right := resources[i]
-		if i+1 < len(resources) {
-			right = resources[i+1]
+// MergeTestShards merge a set of CachedShard by shardKey
+func MergeTestShards(shards []*CachedShard) []*CachedShard {
+	results := make([]*CachedShard, 0, len(shards)/2)
+	for i := 0; i < len(shards); i += 2 {
+		left := shards[i]
+		right := shards[i]
+		if i+1 < len(shards) {
+			right = shards[i+1]
 		}
 
 		leftStart, _ := left.Meta.GetRange()
 		_, rightEnd := right.Meta.GetRange()
 		res := &CachedShard{
 			Meta: metapb.Shard{
-				ID:    left.Meta.GetID() + uint64(len(resources)),
+				ID:    left.Meta.GetID() + uint64(len(shards)),
 				Start: leftStart,
 				End:   rightEnd,
 			},
@@ -96,8 +96,8 @@ func NewTestCachedShard(start, end []byte) *CachedShard {
 	}
 }
 
-// NewTestStoreInfoWithLabel is create a container with specified labels.
-func NewTestStoreInfoWithLabel(id uint64, resourceCount int, labels map[string]string) *CachedStore {
+// NewTestStoreInfoWithLabel creates a store with specified labels.
+func NewTestStoreInfoWithLabel(id uint64, shardCount int, labels map[string]string) *CachedStore {
 	containerLabels := make([]metapb.Label, 0, len(labels))
 	for k, v := range labels {
 		containerLabels = append(containerLabels, metapb.Label{
@@ -111,30 +111,14 @@ func NewTestStoreInfoWithLabel(id uint64, resourceCount int, labels map[string]s
 	container := NewCachedStore(
 		metapb.Store{ID: id, Labels: containerLabels},
 		SetStoreStats(stats),
-		SetShardCount("", resourceCount),
-		SetShardSize("", int64(resourceCount)*10),
+		SetShardCount("", shardCount),
+		SetShardSize("", int64(shardCount)*10),
 	)
 	return container
 }
 
-// NewTestCachedStoreWithSizeCount is create a container with size and count.
-func NewTestCachedStoreWithSizeCount(id uint64, resourceCount, leaderCount int, resourceSize, leaderSize int64) *CachedStore {
-	stats := &metapb.StoreStats{}
-	stats.Capacity = uint64(1024)
-	stats.Available = uint64(1024)
-	container := NewCachedStore(
-		metapb.Store{ID: id},
-		SetStoreStats(stats),
-		SetShardCount("", resourceCount),
-		SetShardSize("", resourceSize),
-		SetLeaderCount("", leaderCount),
-		SetLeaderSize("", leaderSize),
-	)
-	return container
-}
-
-func newTestShardItem(start, end []byte) *resourceItem {
-	return &resourceItem{res: NewTestCachedShard(start, end)}
+func newTestShardItem(start, end []byte) *shardItem {
+	return &shardItem{shard: NewTestCachedShard(start, end)}
 }
 
 func newShardWithStat(start, end string, size, keys int64) *CachedShard {
diff --git a/components/prophet/schedule/checker/rule_checker_test.go b/components/prophet/schedule/checker/rule_checker_test.go
index d51bc6e..ad8dfdc 100644
--- a/components/prophet/schedule/checker/rule_checker_test.go
+++ b/components/prophet/schedule/checker/rule_checker_test.go
@@ -455,7 +455,7 @@ func TestIssue3293(t *testing.T) {
 		},
 	})
 	assert.NoError(t, err)
-	s.cluster.DeleteStore(s.cluster.TakeStore(5))
+	s.cluster.DeleteStore(s.cluster.GetStore(5))
 	err = s.ruleManager.SetRule(&placement.Rule{
 		GroupID: "DDL_51",
 		ID:      "default",
diff --git a/components/prophet/storage/kv_etcd_test.go b/components/prophet/storage/kv_etcd_test.go
index 65e5282..c1dcb2c 100644
--- a/components/prophet/storage/kv_etcd_test.go
+++ b/components/prophet/storage/kv_etcd_test.go
@@ -20,7 +20,6 @@ import (
 
 	"github.com/matrixorigin/matrixcube/components/prophet/election"
 	"github.com/matrixorigin/matrixcube/components/prophet/mock"
-	"github.com/matrixorigin/matrixcube/components/prophet/util"
 	"github.com/stretchr/testify/assert"
 )
 
@@ -80,13 +79,6 @@ func TestEtcdKV(t *testing.T) {
 	assert.Equal(t, 2, len(keys))
 	assert.Equal(t, 2, len(values))
 
-	cnt, err := kv.CountRange(key1, key3)
-	assert.NoError(t, err)
-	assert.Equal(t, uint64(2), cnt)
-	cnt, err = kv.CountRange("key/", util.GetPrefixRangeEnd("key/"))
-	assert.NoError(t, err)
-	assert.Equal(t, uint64(3), cnt)
-
 	kv.Remove(key1)
 	kv.Remove(key2)
 	kv.Remove(key3)
@@ -119,12 +111,6 @@ func TestEtcdKV(t *testing.T) {
 	assert.Empty(t, value)
 
 	kv.Save(key1, key1)
-	ok, err = kv.RemoveIfValueMatched(key1, key2)
-	assert.NoError(t, err)
-	assert.False(t, ok)
-	ok, err = kv.RemoveIfValueMatched(key1, key1)
-	assert.NoError(t, err)
-	assert.True(t, ok)
 	value, err = kv.Load(key1)
 	assert.NoError(t, err)
 	assert.Empty(t, value)
diff --git a/components/prophet/storage/storage_test.go b/components/prophet/storage/storage_test.go
index 424b77e..58612e0 100644
--- a/components/prophet/storage/storage_test.go
+++ b/components/prophet/storage/storage_test.go
@@ -253,88 +253,3 @@ func TestPutAndDeleteAndLoadJobs(t *testing.T) {
 	}))
 	assert.Equal(t, 0, c)
 }
-
-func TestPutAndDeleteAndLoadCustomData(t *testing.T) {
-	stopC, port := mock.StartTestSingleEtcd(t)
-	defer close(stopC)
-
-	client := mock.NewEtcdClient(t, port)
-	defer client.Close()
-
-	e, err := election.NewElector(client)
-	assert.NoError(t, err)
-	ls := e.CreateLeadship("prophet", "node1", "node1", true, func(string) bool { return true }, func(string) bool { return true })
-	defer ls.Stop()
-
-	ls.ElectionLoop()
-	time.Sleep(time.Millisecond * 200)
-
-	rootPath := "/root"
-	storage := NewStorage(
-		rootPath,
-		NewEtcdKV("/root", client, ls),
-		id.NewEtcdGenerator(rootPath, client, ls),
-	)
-
-	assert.NoError(t, storage.PutCustomData([]byte("k1"), []byte("v1")))
-	assert.NoError(t, storage.PutCustomData([]byte("k2"), []byte("v2")))
-	assert.NoError(t, storage.PutCustomData([]byte("k3"), []byte("v3")))
-
-	var loadedValues [][]byte
-	assert.NoError(t, storage.LoadCustomData(1, func(k, v []byte) error {
-		loadedValues = append(loadedValues, k)
-		return nil
-	}))
-	assert.Equal(t, 3, len(loadedValues))
-	assert.Equal(t, []byte("k1"), loadedValues[0])
-	assert.NoError(t, storage.RemoveCustomData(loadedValues[0]))
-	assert.Equal(t, []byte("k2"), loadedValues[1])
-	assert.NoError(t, storage.RemoveCustomData(loadedValues[1]))
-	assert.Equal(t, []byte("k3"), loadedValues[2])
-	assert.NoError(t, storage.RemoveCustomData(loadedValues[2]))
-
-	c := 0
-	assert.NoError(t, storage.LoadCustomData(1, func(k, v []byte) error {
-		c++
-		return nil
-	}))
-	assert.Equal(t, 0, c)
-}
-
-func TestBatchPutCustomData(t *testing.T) {
-	stopC, port := mock.StartTestSingleEtcd(t)
-	defer close(stopC)
-
-	client := mock.NewEtcdClient(t, port)
-	defer client.Close()
-
-	e, err := election.NewElector(client)
-	assert.NoError(t, err)
-	ls := e.CreateLeadship("prophet", "node1", "node1", true, func(string) bool { return true }, func(string) bool { return true })
-	defer ls.Stop()
-
-	ls.ElectionLoop()
-	time.Sleep(time.Millisecond * 200)
-
-	keys := [][]byte{[]byte("k1"), []byte("k2"), []byte("k3")}
-	data := [][]byte{[]byte("v1"), []byte("v2"), []byte("v3")}
-
-	rootPath := "/root"
-	storage := NewStorage(
-		rootPath,
-		NewEtcdKV("/root", client, ls),
-		id.NewEtcdGenerator(rootPath, client, ls),
-	)
-
-	assert.NoError(t, storage.BatchPutCustomData(keys, data))
-
-	var loadedValues [][]byte
-	assert.NoError(t, storage.LoadCustomData(1, func(k, v []byte) error {
-		loadedValues = append(loadedValues, v)
-		return nil
-	}))
-	assert.Equal(t, len(keys), len(loadedValues))
-	for i := 0; i < len(keys); i++ {
-		assert.Equal(t, data[i], loadedValues[i])
-	}
-}
diff --git a/raftstore/integration_create_shard_test.go b/raftstore/integration_create_shard_test.go
index 3a00c97..5d04e5c 100644
--- a/raftstore/integration_create_shard_test.go
+++ b/raftstore/integration_create_shard_test.go
@@ -33,8 +33,8 @@ type testShardsAware struct {
 	adjust func(*core.CachedShard) *core.CachedShard
 }
 
-func (tra *testShardsAware) ForeachWaittingCreateShards(do func(res metapb.Shard)) {
-	tra.aware.ForeachWaittingCreateShards(do)
+func (tra *testShardsAware) ForeachWaitingCreateShards(do func(res metapb.Shard)) {
+	tra.aware.ForeachWaitingCreateShards(do)
 }
 func (tra *testShardsAware) ForeachShards(group uint64, fn func(res metapb.Shard)) {
 	tra.aware.ForeachShards(group, fn)
diff --git a/raftstore/store_shards_pool_test.go b/raftstore/store_shards_pool_test.go
index 7898e54..d1756aa 100644
--- a/raftstore/store_shards_pool_test.go
+++ b/raftstore/store_shards_pool_test.go
@@ -267,7 +267,7 @@ func TestDoAllocLocked(t *testing.T) {
 	// pd has no corresponding data
 	aware = mockjob.NewMockShardsAware(ctrl)
 	aware.EXPECT().ForeachShards(gomock.Any(), gomock.Any()).DoAndReturn(func(group uint64, fn func(res metapb.Shard)) {})
-	aware.EXPECT().ForeachWaittingCreateShards(gomock.Any()).DoAndReturn(func(fn func(res metapb.Shard)) {})
+	aware.EXPECT().ForeachWaitingCreateShards(gomock.Any()).DoAndReturn(func(fn func(res metapb.Shard)) {})
 	p.mu.createC = make(chan struct{}, 10)
 	p.mu.pools = metapb.ShardsPool{Pools: make(map[uint64]*metapb.ShardPool)}
 	p.mu.pools.Pools[0] = &metapb.ShardPool{Capacity: 1, Seq: 1, AllocatedOffset: 0}
@@ -279,7 +279,7 @@ func TestDoAllocLocked(t *testing.T) {
 
 	// allocate from created
 	aware = mockjob.NewMockShardsAware(ctrl)
-	aware.EXPECT().ForeachWaittingCreateShards(gomock.Any()).DoAndReturn(func(fn func(res metapb.Shard)) {
+	aware.EXPECT().ForeachWaitingCreateShards(gomock.Any()).DoAndReturn(func(fn func(res metapb.Shard)) {
 		fn(metapb.Shard{ID: 1, Unique: p.unique(0, 1)})
 	})
 	p.mu.createC = make(chan struct{}, 10)