milvus-io
/
milvus
mirror of https://github.com/milvus-io/milvus.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Support Range Query 

Signed-off-by: FluorineDog <guilin.gou@zilliz.com>
pull/4973/head^2
FluorineDog 2020-11-23 11:56:45 +08:00 committed by yefu.chen
parent 3277987898
commit 76001b29b6
6 changed files with 66 additions and 85 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

41
cmd/master/main.go
View File

@ -29,27 +29,26 @@ func main() {
statsChannel := master.Params.StatsChannels() statsChannel := master.Params.StatsChannels()
opt := master.Option{ opt := master.Option{
KVRootPath: etcdRootPath, KVRootPath: etcdRootPath,
MetaRootPath: etcdRootPath, MetaRootPath: etcdRootPath,
EtcdAddr: []string{etcdAddress}, EtcdAddr: []string{etcdAddress},
PulsarAddr: pulsarAddr, PulsarAddr: pulsarAddr,
ProxyIDs: master.Params.ProxyIDList(), ProxyIDs: master.Params.ProxyIDList(),
PulsarProxyChannels: master.Params.ProxyTimeSyncChannels(), PulsarProxyChannels: master.Params.ProxyTimeSyncChannels(),
PulsarProxySubName: master.Params.ProxyTimeSyncSubName(), PulsarProxySubName: master.Params.ProxyTimeSyncSubName(),
SoftTTBInterval: master.Params.SoftTimeTickBarrierInterval(), SoftTTBInterval: master.Params.SoftTimeTickBarrierInterval(),
WriteIDs: master.Params.WriteIDList(), WriteIDs: master.Params.WriteIDList(),
PulsarWriteChannels: master.Params.WriteTimeSyncChannels(), PulsarWriteChannels: master.Params.WriteTimeSyncChannels(),
PulsarWriteSubName: master.Params.WriteTimeSyncSubName(), PulsarWriteSubName: master.Params.WriteTimeSyncSubName(),
PulsarDMChannels: master.Params.DMTimeSyncChannels(), PulsarDMChannels: master.Params.DMTimeSyncChannels(),
PulsarK2SChannels: master.Params.K2STimeSyncChannels(), PulsarK2SChannels: master.Params.K2STimeSyncChannels(),
DefaultRecordSize: defaultRecordSize, DefaultRecordSize: defaultRecordSize,
MinimumAssignSize: minimumAssignSize, MinimumAssignSize: minimumAssignSize,
SegmentThreshold: segmentThreshold, SegmentThreshold: segmentThreshold,
SegmentThresholdFactor: master.Params.SegmentThresholdFactor(), SegmentExpireDuration: segmentExpireDuration,
SegmentExpireDuration: segmentExpireDuration, NumOfChannel: numOfChannel,
NumOfChannel: numOfChannel, NumOfQueryNode: nodeNum,
NumOfQueryNode: nodeNum, StatsChannels: statsChannel,
StatsChannels: statsChannel,
} }
svr, err := master.CreateServer(ctx, &opt) svr, err := master.CreateServer(ctx, &opt)

1
configs/config.yaml
View File

@ -31,7 +31,6 @@ master:
minimumAssignSize: 1048576 minimumAssignSize: 1048576
segmentThreshold: 536870912 segmentThreshold: 536870912
segmentExpireDuration: 2000 segmentExpireDuration: 2000
segmentThresholdFactor: 0.75
querynodenum: 1 querynodenum: 1
writenodenum: 1 writenodenum: 1
statsChannels: "statistic" statsChannels: "statistic"

15
internal/master/master.go
View File

@ -48,14 +48,13 @@ type Option struct {
PulsarDMChannels []string PulsarDMChannels []string
PulsarK2SChannels []string PulsarK2SChannels []string
DefaultRecordSize int64 DefaultRecordSize int64
MinimumAssignSize int64 MinimumAssignSize int64
SegmentThreshold float64 SegmentThreshold float64
SegmentThresholdFactor float64 SegmentExpireDuration int64
SegmentExpireDuration int64 NumOfChannel int
NumOfChannel int NumOfQueryNode int
NumOfQueryNode int StatsChannels string
StatsChannels string
} }
type Master struct { type Master struct {

12
internal/master/paramtable.go
View File

@ -214,15 +214,3 @@ func (p *ParamTable) TopicNum() int {
} }
return num return num
} }
func (p *ParamTable) SegmentThresholdFactor() float64 {
factor, err := p.Load("master.segmentThresholdFactor")
if err != nil {
panic(err)
}
res, err := strconv.ParseFloat(factor, 64)
if err != nil {
panic(err)
}
return res
}

65
internal/master/segment_manager.go
View File

@ -29,21 +29,20 @@ type segmentStatus struct {
} }
type SegmentManager struct { type SegmentManager struct {
metaTable *metaTable metaTable *metaTable
statsStream msgstream.MsgStream statsStream msgstream.MsgStream
channelRanges []*channelRange channelRanges []*channelRange
segmentStatus map[UniqueID]*segmentStatus // segment id to segment status segmentStatus map[UniqueID]*segmentStatus // segment id to segment status
collStatus map[UniqueID]*collectionStatus // collection id to collection status collStatus map[UniqueID]*collectionStatus // collection id to collection status
defaultSizePerRecord int64 defaultSizePerRecord int64
minimumAssignSize int64 minimumAssignSize int64
segmentThreshold float64 segmentThreshold int64
segmentThresholdFactor float64 segmentExpireDuration int64
segmentExpireDuration int64 numOfChannels int
numOfChannels int numOfQueryNodes int
numOfQueryNodes int globalIDAllocator func() (UniqueID, error)
globalIDAllocator func() (UniqueID, error) globalTSOAllocator func() (Timestamp, error)
globalTSOAllocator func() (Timestamp, error) mu sync.RWMutex
mu sync.RWMutex
} }
func (segMgr *SegmentManager) HandleQueryNodeMsgPack(msgPack *msgstream.MsgPack) error { func (segMgr *SegmentManager) HandleQueryNodeMsgPack(msgPack *msgstream.MsgPack) error {
@ -77,7 +76,7 @@ func (segMgr *SegmentManager) handleSegmentStat(segStats *internalpb.SegmentStat
segMeta.NumRows = segStats.NumRows segMeta.NumRows = segStats.NumRows
segMeta.MemSize = segStats.MemorySize segMeta.MemSize = segStats.MemorySize
if segStats.MemorySize > int64(segMgr.segmentThresholdFactor*segMgr.segmentThreshold) { if segStats.MemorySize > segMgr.segmentThreshold {
return segMgr.closeSegment(segMeta) return segMgr.closeSegment(segMeta)
} }
return segMgr.metaTable.UpdateSegment(segMeta) return segMgr.metaTable.UpdateSegment(segMeta)
@ -151,7 +150,6 @@ func (segMgr *SegmentManager) AssignSegmentID(segIDReq []*internalpb.SegIDReques
func (segMgr *SegmentManager) assignSegment(collName string, collID UniqueID, partitionTag string, count uint32, channelID int32, func (segMgr *SegmentManager) assignSegment(collName string, collID UniqueID, partitionTag string, count uint32, channelID int32,
collStatus *collectionStatus) (*internalpb.SegIDAssignment, error) { collStatus *collectionStatus) (*internalpb.SegIDAssignment, error) {
segmentThreshold := int64(segMgr.segmentThreshold)
for _, segID := range collStatus.openedSegments { for _, segID := range collStatus.openedSegments {
segMeta, _ := segMgr.metaTable.GetSegmentByID(segID) segMeta, _ := segMgr.metaTable.GetSegmentByID(segID)
if segMeta.GetCloseTime() != 0 || channelID < segMeta.GetChannelStart() || if segMeta.GetCloseTime() != 0 || channelID < segMeta.GetChannelStart() ||
@ -162,8 +160,8 @@ func (segMgr *SegmentManager) assignSegment(collName string, collID UniqueID, pa
assignedMem := segMgr.checkAssignedSegExpire(segID) assignedMem := segMgr.checkAssignedSegExpire(segID)
memSize := segMeta.MemSize memSize := segMeta.MemSize
neededMemSize := segMgr.calNeededSize(memSize, segMeta.NumRows, int64(count)) neededMemSize := segMgr.calNeededSize(memSize, segMeta.NumRows, int64(count))
if memSize+assignedMem+neededMemSize <= segmentThreshold { if memSize+assignedMem+neededMemSize <= segMgr.segmentThreshold {
remainingSize := segmentThreshold - memSize - assignedMem remainingSize := segMgr.segmentThreshold - memSize - assignedMem
allocMemSize := segMgr.calAllocMemSize(neededMemSize, remainingSize) allocMemSize := segMgr.calAllocMemSize(neededMemSize, remainingSize)
segMgr.addAssignment(segID, allocMemSize) segMgr.addAssignment(segID, allocMemSize)
return &internalpb.SegIDAssignment{ return &internalpb.SegIDAssignment{
@ -176,7 +174,7 @@ func (segMgr *SegmentManager) assignSegment(collName string, collID UniqueID, pa
} }
} }
neededMemSize := segMgr.defaultSizePerRecord * int64(count) neededMemSize := segMgr.defaultSizePerRecord * int64(count)
if neededMemSize > segmentThreshold { if neededMemSize > segMgr.segmentThreshold {
return nil, errors.Errorf("request with count %d need about %d mem size which is larger than segment threshold", return nil, errors.Errorf("request with count %d need about %d mem size which is larger than segment threshold",
count, neededMemSize) count, neededMemSize)
} }
@ -186,7 +184,7 @@ func (segMgr *SegmentManager) assignSegment(collName string, collID UniqueID, pa
return nil, err return nil, err
} }
allocMemSize := segMgr.calAllocMemSize(neededMemSize, segmentThreshold) allocMemSize := segMgr.calAllocMemSize(neededMemSize, segMgr.segmentThreshold)
segMgr.addAssignment(segMeta.SegmentID, allocMemSize) segMgr.addAssignment(segMeta.SegmentID, allocMemSize)
return &internalpb.SegIDAssignment{ return &internalpb.SegIDAssignment{
SegID: segMeta.SegmentID, SegID: segMeta.SegmentID,
@ -329,19 +327,18 @@ func NewSegmentManager(meta *metaTable,
globalTSOAllocator func() (Timestamp, error), globalTSOAllocator func() (Timestamp, error),
) *SegmentManager { ) *SegmentManager {
segMgr := &SegmentManager{ segMgr := &SegmentManager{
metaTable: meta, metaTable: meta,
channelRanges: make([]*channelRange, 0), channelRanges: make([]*channelRange, 0),
segmentStatus: make(map[UniqueID]*segmentStatus), segmentStatus: make(map[UniqueID]*segmentStatus),
collStatus: make(map[UniqueID]*collectionStatus), collStatus: make(map[UniqueID]*collectionStatus),
segmentThreshold: opt.SegmentThreshold, segmentThreshold: int64(opt.SegmentThreshold),
segmentThresholdFactor: opt.SegmentThresholdFactor, segmentExpireDuration: opt.SegmentExpireDuration,
segmentExpireDuration: opt.SegmentExpireDuration, minimumAssignSize: opt.MinimumAssignSize,
minimumAssignSize: opt.MinimumAssignSize, defaultSizePerRecord: opt.DefaultRecordSize,
defaultSizePerRecord: opt.DefaultRecordSize, numOfChannels: opt.NumOfChannel,
numOfChannels: opt.NumOfChannel, numOfQueryNodes: opt.NumOfQueryNode,
numOfQueryNodes: opt.NumOfQueryNode, globalIDAllocator: globalIDAllocator,
globalIDAllocator: globalIDAllocator, globalTSOAllocator: globalTSOAllocator,
globalTSOAllocator: globalTSOAllocator,
} }
segMgr.createChannelRanges() segMgr.createChannelRanges()
return segMgr return segMgr

17
internal/master/segment_manager_test.go
View File

@ -7,7 +7,7 @@ import (
"testing" "testing"
"time" "time"
"github.com/golang/protobuf/proto" "github.com/gogo/protobuf/proto"
"github.com/zilliztech/milvus-distributed/internal/proto/commonpb" "github.com/zilliztech/milvus-distributed/internal/proto/commonpb"
"github.com/zilliztech/milvus-distributed/internal/proto/masterpb" "github.com/zilliztech/milvus-distributed/internal/proto/masterpb"
"github.com/zilliztech/milvus-distributed/internal/proto/servicepb" "github.com/zilliztech/milvus-distributed/internal/proto/servicepb"
@ -78,13 +78,12 @@ func setup() {
panic(err) panic(err)
} }
opt := &Option{ opt := &Option{
SegmentThreshold: 536870912, SegmentThreshold: 536870912,
SegmentExpireDuration: 2000, SegmentExpireDuration: 2000,
MinimumAssignSize: 1048576, MinimumAssignSize: 1048576,
DefaultRecordSize: 1024, DefaultRecordSize: 1024,
NumOfQueryNode: 3, NumOfQueryNode: 3,
NumOfChannel: 5, NumOfChannel: 5,
SegmentThresholdFactor: 0.75,
} }
var cnt int64 var cnt int64
@ -210,7 +209,7 @@ func TestSegmentManager_SegmentStats(t *testing.T) {
// close segment // close segment
stats.SegStats[0].NumRows = 600000 stats.SegStats[0].NumRows = 600000
stats.SegStats[0].MemorySize = int64(0.8 * segMgr.segmentThreshold) stats.SegStats[0].MemorySize = 600000000
err = segMgr.HandleQueryNodeMsgPack(&msgPack) err = segMgr.HandleQueryNodeMsgPack(&msgPack)
assert.Nil(t, err) assert.Nil(t, err)
segMeta, _ = mt.GetSegmentByID(100) segMeta, _ = mt.GetSegmentByID(100)