Inforschema, executor: add push down for tikv_region_peers

executor/builder.go

@@ -1607,6 +1607,14 @@ func (b *executorBuilder) buildMemTable(v *plannercore.PhysicalMemTable) Executo
 					timeRange: v.QueryTimeRange,
 				},
 			}
+		case strings.ToLower(infoschema.TableTiKVRegionPeers):
+			return &MemTableReaderExec{
+				baseExecutor: newBaseExecutor(b.ctx, v.Schema(), v.ID()),
+				table:        v.Table,
+				retriever: &tikvRegionPeersRetriever{
+					extractor: v.Extractor.(*plannercore.TikvRegionPeersExtractor),
+				},
+			}
 		case strings.ToLower(infoschema.TableSchemata),
 			strings.ToLower(infoschema.TableStatistics),
 			strings.ToLower(infoschema.TableTiDBIndexes),
@@ -1628,7 +1636,6 @@ func (b *executorBuilder) buildMemTable(v *plannercore.PhysicalMemTable) Executo
 			strings.ToLower(infoschema.TableProcesslist),
 			strings.ToLower(infoschema.ClusterTableProcesslist),
 			strings.ToLower(infoschema.TableTiKVRegionStatus),
-			strings.ToLower(infoschema.TableTiKVRegionPeers),
 			strings.ToLower(infoschema.TableTiDBHotRegions),
 			strings.ToLower(infoschema.TableSessionVar),
 			strings.ToLower(infoschema.TableConstraints),

executor/infoschema_reader.go

@@ -141,8 +141,6 @@ func (e *memtableRetriever) retrieve(ctx context.Context, sctx sessionctx.Contex
 			e.setDataFromUserPrivileges(sctx)
 		case infoschema.TableTiKVRegionStatus:
 			err = e.setDataForTiKVRegionStatus(sctx)
-		case infoschema.TableTiKVRegionPeers:
-			err = e.setDataForTikVRegionPeers(sctx)
 		case infoschema.TableTiDBHotRegions:
 			err = e.setDataForTiDBHotRegions(sctx)
 		case infoschema.TableConstraints:
@@ -1495,63 +1493,6 @@ func (e *memtableRetriever) setNewTiKVRegionStatusCol(region *helper.RegionInfo,
 	e.rows = append(e.rows, row)
 }
 
-func (e *memtableRetriever) setDataForTikVRegionPeers(ctx sessionctx.Context) error {
-	tikvStore, ok := ctx.GetStore().(helper.Storage)
-	if !ok {
-		return errors.New("Information about TiKV region status can be gotten only when the storage is TiKV")
-	}
-	tikvHelper := &helper.Helper{
-		Store:       tikvStore,
-		RegionCache: tikvStore.GetRegionCache(),
-	}
-	regionsInfo, err := tikvHelper.GetRegionsInfo()
-	if err != nil {
-		return err
-	}
-	for i := range regionsInfo.Regions {
-		e.setNewTiKVRegionPeersCols(&regionsInfo.Regions[i])
-	}
-	return nil
-}
-
-func (e *memtableRetriever) setNewTiKVRegionPeersCols(region *helper.RegionInfo) {
-	records := make([][]types.Datum, 0, len(region.Peers))
-	pendingPeerIDSet := set.NewInt64Set()
-	for _, peer := range region.PendingPeers {
-		pendingPeerIDSet.Insert(peer.ID)
-	}
-	downPeerMap := make(map[int64]int64, len(region.DownPeers))
-	for _, peerStat := range region.DownPeers {
-		downPeerMap[peerStat.Peer.ID] = peerStat.DownSec
-	}
-	for _, peer := range region.Peers {
-		row := make([]types.Datum, len(infoschema.TableTiKVRegionPeersCols))
-		row[0].SetInt64(region.ID)
-		row[1].SetInt64(peer.ID)
-		row[2].SetInt64(peer.StoreID)
-		if peer.IsLearner {
-			row[3].SetInt64(1)
-		} else {
-			row[3].SetInt64(0)
-		}
-		if peer.ID == region.Leader.ID {
-			row[4].SetInt64(1)
-		} else {
-			row[4].SetInt64(0)
-		}
-		if downSec, ok := downPeerMap[peer.ID]; ok {
-			row[5].SetString(downPeer, mysql.DefaultCollationName)
-			row[6].SetInt64(downSec)
-		} else if pendingPeerIDSet.Exist(peer.ID) {
-			row[5].SetString(pendingPeer, mysql.DefaultCollationName)
-		} else {
-			row[5].SetString(normalPeer, mysql.DefaultCollationName)
-		}
-		records = append(records, row)
-	}
-	e.rows = append(e.rows, records...)
-}
-
 const (
 	normalPeer  = "NORMAL"
 	pendingPeer = "PENDING"

executor/memtable_reader.go

@@ -997,3 +997,116 @@ func (e *hotRegionsHistoryRetriver) getHotRegionRowWithSchemaInfo(
 	}
 	return row, nil
 }
+
+type tikvRegionPeersRetriever struct {
+	dummyCloser
+	extractor *plannercore.TikvRegionPeersExtractor
+	retrieved bool
+}
+
+func (e *tikvRegionPeersRetriever) retrieve(ctx context.Context, sctx sessionctx.Context) ([][]types.Datum, error) {
+	if e.extractor.SkipRequest || e.retrieved {
+		return nil, nil
+	}
+	e.retrieved = true
+	tikvStore, ok := sctx.GetStore().(helper.Storage)
+	if !ok {
+		return nil, errors.New("Information about hot region can be gotten only when the storage is TiKV")
+	}
+	tikvHelper := &helper.Helper{
+		Store:       tikvStore,
+		RegionCache: tikvStore.GetRegionCache(),
+	}
+
+	if len(e.extractor.StoreIDs) < 1 && len(e.extractor.RegionIDs) < 1 {
+		regionsInfo, err := tikvHelper.GetRegionsInfo()
+		if err != nil {
+			return nil, err
+		}
+		return e.packTiKVRegionPeersRows(regionsInfo.Regions)
+	}
+
+	var regionsInfo, regionsInfoByStoreID, regionsInfoByRegionID []helper.RegionInfo
+	regionMap := make(map[int64]struct{})
+	if len(e.extractor.StoreIDs) > 0 {
+		for _, storeID := range e.extractor.StoreIDs {
+			regionsInfo, err := tikvHelper.GetStoreRegionsInfo(storeID)
+			if err != nil {
+				return nil, err
+			}
+			// remove dup region
+			for _, region := range regionsInfo.Regions {
+				if _, ok := regionMap[region.ID]; !ok {
+					regionMap[region.ID] = struct{}{}
+					regionsInfoByStoreID = append(regionsInfoByStoreID, region)
+				}
+			}
+		}
+		if len(e.extractor.RegionIDs) < 1 {
+			return e.packTiKVRegionPeersRows(regionsInfoByStoreID)
+		}
+	}
+
+	if len(e.extractor.RegionIDs) > 0 {
+		for _, regionID := range e.extractor.RegionIDs {
+			regionInfo, err := tikvHelper.GetRegionInfoByID(regionID)
+			if err != nil {
+				return nil, err
+			}
+			regionsInfoByRegionID = append(regionsInfoByRegionID, *regionInfo)
+		}
+		if len(e.extractor.StoreIDs) < 1 {
+			return e.packTiKVRegionPeersRows(regionsInfoByRegionID)
+		}
+	}
+
+	// intersect
+	for _, region := range regionsInfoByRegionID {
+		if _, ok := regionMap[region.ID]; ok {
+			regionsInfo = append(regionsInfo, region)
+		}
+	}
+	return e.packTiKVRegionPeersRows(regionsInfo)
+}
+
+func (e *tikvRegionPeersRetriever) packTiKVRegionPeersRows(regionsInfo []helper.RegionInfo) ([][]types.Datum, error) {
+	var rows [][]types.Datum
+	for _, region := range regionsInfo {
+		records := make([][]types.Datum, 0, len(region.Peers))
+		pendingPeerIDSet := set.NewInt64Set()
+		for _, peer := range region.PendingPeers {
+			pendingPeerIDSet.Insert(peer.ID)
+		}
+		downPeerMap := make(map[int64]int64, len(region.DownPeers))
+		for _, peerStat := range region.DownPeers {
+			downPeerMap[peerStat.Peer.ID] = peerStat.DownSec
+		}
+		for _, peer := range region.Peers {
+			row := make([]types.Datum, len(infoschema.TableTiKVRegionPeersCols))
+			row[0].SetInt64(region.ID)
+			row[1].SetInt64(peer.ID)
+			row[2].SetInt64(peer.StoreID)
+			if peer.IsLearner {
+				row[3].SetInt64(1)
+			} else {
+				row[3].SetInt64(0)
+			}
+			if peer.ID == region.Leader.ID {
+				row[4].SetInt64(1)
+			} else {
+				row[4].SetInt64(0)
+			}
+			if downSec, ok := downPeerMap[peer.ID]; ok {
+				row[5].SetString(downPeer, mysql.DefaultCollationName)
+				row[6].SetInt64(downSec)
+			} else if pendingPeerIDSet.Exist(peer.ID) {
+				row[5].SetString(pendingPeer, mysql.DefaultCollationName)
+			} else {
+				row[5].SetString(normalPeer, mysql.DefaultCollationName)
+			}
+			records = append(records, row)
+		}
+		rows = append(rows, records...)
+	}
+	return rows, nil
+}