region_sync: region syncer support bucket

server/region_syncer/client.go

@@ -186,8 +186,10 @@ func (s *RegionSyncer) StartSyncWithLeader(addr string) {
  }
  stats := resp.GetRegionStats()
  regions := resp.GetRegions()
+ buckets := resp.GetBuckets()
  regionLeaders := resp.GetRegionLeaders()
  hasStats := len(stats) == len(regions)
+ hasBuckets := len(buckets) == len(regions)
  for i, r := range regions {
  var (
  region *core.RegionInfo
@@ -207,6 +209,10 @@ func (s *RegionSyncer) StartSyncWithLeader(addr string) {
  region = core.NewRegionInfo(r, regionLeader)
  }
 
+ if hasBuckets {
+ region.UpdateBuckets(buckets[i], region.GetBuckets())
+ }
+
  origin, err := bc.PreCheckPutRegion(region)
  if err != nil {
  log.Debug("region is stale", zap.Stringer("origin", origin.GetMeta()), errs.ZapError(err))

server/region_syncer/server.go

@@ -113,6 +113,7 @@ func (s *RegionSyncer) RunServer(ctx context.Context, regionNotifier <-chan *cor
  var requests []*metapb.Region
  var stats []*pdpb.RegionStat
  var leaders []*metapb.Peer
+ var buckets []*metapb.Buckets
  ticker := time.NewTicker(syncerKeepAliveInterval)
 
  defer func() {
@@ -130,6 +131,7 @@ func (s *RegionSyncer) RunServer(ctx context.Context, regionNotifier <-chan *cor
  case first := <-regionNotifier:
  requests = append(requests, first.GetMeta())
  stats = append(stats, first.GetStat())
+ buckets = append(buckets, first.GetBuckets())
  leaders = append(leaders, first.GetLeader())
  startIndex := s.history.GetNextIndex()
  s.history.Record(first)
@@ -138,6 +140,7 @@ func (s *RegionSyncer) RunServer(ctx context.Context, regionNotifier <-chan *cor
  region := <-regionNotifier
  requests = append(requests, region.GetMeta())
  stats = append(stats, region.GetStat())
+ buckets = append(buckets, region.GetBuckets())
  leaders = append(leaders, region.GetLeader())
  s.history.Record(region)
  }
@@ -147,6 +150,7 @@ func (s *RegionSyncer) RunServer(ctx context.Context, regionNotifier <-chan *cor
  StartIndex: startIndex,
  RegionStats: stats,
  RegionLeaders: leaders,
+ Buckets: buckets,
  }
  s.broadcast(regions)
  case <-ticker.C:
@@ -159,6 +163,7 @@ func (s *RegionSyncer) RunServer(ctx context.Context, regionNotifier <-chan *cor
  requests = requests[:0]
  stats = stats[:0]
  leaders = leaders[:0]
+ buckets = buckets[:0]
  }
 }
 
@@ -225,6 +230,7 @@ func (s *RegionSyncer) syncHistoryRegion(ctx context.Context, request *pdpb.Sync
  metas := make([]*metapb.Region, 0, maxSyncRegionBatchSize)
  stats := make([]*pdpb.RegionStat, 0, maxSyncRegionBatchSize)
  leaders := make([]*metapb.Peer, 0, maxSyncRegionBatchSize)
+ buckets := make([]*metapb.Buckets, 0, maxSyncRegionBatchSize)
  for syncedIndex, r := range regions {
  select {
  case <-ctx.Done():
@@ -243,6 +249,11 @@ func (s *RegionSyncer) syncHistoryRegion(ctx context.Context, request *pdpb.Sync
  leader = r.GetLeader()
  }
  leaders = append(leaders, leader)
+ bucket := &metapb.Buckets{}
+ if r.GetBuckets() != nil {
+ bucket = r.GetBuckets()
+ }
+ buckets = append(buckets, bucket)
  if len(metas) < maxSyncRegionBatchSize && syncedIndex < len(regions)-1 {
  continue
  }
@@ -252,6 +263,7 @@ func (s *RegionSyncer) syncHistoryRegion(ctx context.Context, request *pdpb.Sync
  StartIndex: uint64(lastIndex),
  RegionStats: stats,
  RegionLeaders: leaders,
+ Buckets: buckets,
  }
  s.limit.WaitN(ctx, resp.Size())
  lastIndex += len(metas)
@@ -262,6 +274,7 @@ func (s *RegionSyncer) syncHistoryRegion(ctx context.Context, request *pdpb.Sync
  metas = metas[:0]
  stats = stats[:0]
  leaders = leaders[:0]
+ buckets = buckets[:0]
  }
  log.Info("requested server has completed full synchronization with server",
  zap.String("requested-server", name), zap.String("server", s.server.Name()), zap.Duration("cost", time.Since(start)))
@@ -278,6 +291,7 @@ func (s *RegionSyncer) syncHistoryRegion(ctx context.Context, request *pdpb.Sync
  regions := make([]*metapb.Region, len(records))
  stats := make([]*pdpb.RegionStat, len(records))
  leaders := make([]*metapb.Peer, len(records))
+ buckets := make([]*metapb.Buckets, len(records))
  for i, r := range records {
  regions[i] = r.GetMeta()
  stats[i] = r.GetStat()
@@ -286,13 +300,15 @@ func (s *RegionSyncer) syncHistoryRegion(ctx context.Context, request *pdpb.Sync
  leader = r.GetLeader()
  }
  leaders[i] = leader
+ buckets[i] = r.GetBuckets()
  }
  resp := &pdpb.SyncRegionResponse{
  Header: &pdpb.ResponseHeader{ClusterId: s.server.ClusterID()},
  Regions: regions,
  StartIndex: startIndex,
  RegionStats: stats,
  RegionLeaders: leaders,
+ Buckets: buckets,
  }
  return stream.Send(resp)
 }

tests/server/region_syncer/region_syncer_test.go

@@ -164,6 +164,8 @@ func (s *regionSyncerTestSuite) TestRegionSyncer(c *C) {
  c.Assert(r.GetMeta(), DeepEquals, region.GetMeta())
  c.Assert(r.GetStat(), DeepEquals, region.GetStat())
  c.Assert(r.GetLeader(), DeepEquals, region.GetLeader())
+ c.Assert(r.GetBuckets().GetKeys(), HasLen, 2)
+ c.Assert(r.GetBuckets(), DeepEquals, region.GetBuckets())
  }
 }
 
@@ -226,7 +228,14 @@ func initRegions(regionLen int) []*core.RegionInfo {
  {Id: allocator.alloc(), StoreId: uint64(0)},
  },
  }
- regions = append(regions, core.NewRegionInfo(r, r.Peers[0]))
+ buckets := &metapb.Buckets{
+ RegionId: r.Id,
+ Keys: [][]byte{r.StartKey, r.EndKey},
+ Version: 1,
+ }
+ region := core.NewRegionInfo(r, r.Peers[0])
+ region.UpdateBuckets(buckets, region.GetBuckets())
+ regions = append(regions, region)
  }
  return regions
 }