@@ -23,17 +23,15 @@ import (
2323 "sync"
2424 "time"
2525
26+ "github.com/samber/lo"
27+ "go.uber.org/zap"
28+
2629 "github.com/milvus-io/milvus-proto/go-api/commonpb"
2730 "github.com/milvus-io/milvus/internal/log"
2831 "github.com/milvus-io/milvus/internal/proto/datapb"
29- "github.com/milvus-io/milvus/internal/proto/indexpb"
30- "github.com/milvus-io/milvus/internal/types"
31- "github.com/milvus-io/milvus/internal/util/funcutil"
3232 "github.com/milvus-io/milvus/internal/util/indexparamcheck"
3333 "github.com/milvus-io/milvus/internal/util/logutil"
3434 "github.com/milvus-io/milvus/internal/util/tsoutil"
35- "github.com/samber/lo"
36- "go.uber.org/zap"
3735)
3836
3937type compactTime struct {
@@ -66,17 +64,17 @@ type compactionSignal struct {
6664var _ trigger = (* compactionTrigger )(nil )
6765
6866type compactionTrigger struct {
69- handler Handler
70- meta * meta
71- allocator allocator
72- signals chan * compactionSignal
73- compactionHandler compactionPlanContext
74- globalTrigger * time.Ticker
75- forceMu sync.Mutex
76- quit chan struct {}
77- wg sync.WaitGroup
78- segRefer * SegmentReferenceManager
79- indexCoord types.IndexCoord
67+ handler Handler
68+ meta * meta
69+ allocator allocator
70+ signals chan * compactionSignal
71+ compactionHandler compactionPlanContext
72+ globalTrigger * time.Ticker
73+ forceMu sync.Mutex
74+ quit chan struct {}
75+ wg sync.WaitGroup
76+ // segRefer *SegmentReferenceManager
77+ // indexCoord types.IndexCoord
8078 estimateNonDiskSegmentPolicy calUpperLimitPolicy
8179 estimateDiskSegmentPolicy calUpperLimitPolicy
8280 // A sloopy hack, so we can test with different segment row count without worrying that
@@ -88,17 +86,17 @@ func newCompactionTrigger(
8886 meta * meta ,
8987 compactionHandler compactionPlanContext ,
9088 allocator allocator ,
91- segRefer * SegmentReferenceManager ,
92- indexCoord types.IndexCoord ,
89+ // segRefer *SegmentReferenceManager,
90+ // indexCoord types.IndexCoord,
9391 handler Handler ,
9492) * compactionTrigger {
9593 return & compactionTrigger {
96- meta : meta ,
97- allocator : allocator ,
98- signals : make (chan * compactionSignal , 100 ),
99- compactionHandler : compactionHandler ,
100- segRefer : segRefer ,
101- indexCoord : indexCoord ,
94+ meta : meta ,
95+ allocator : allocator ,
96+ signals : make (chan * compactionSignal , 100 ),
97+ compactionHandler : compactionHandler ,
98+ // segRefer: segRefer,
99+ // indexCoord: indexCoord,
102100 estimateDiskSegmentPolicy : calBySchemaPolicyWithDiskIndex ,
103101 estimateNonDiskSegmentPolicy : calBySchemaPolicy ,
104102 handler : handler ,
@@ -280,39 +278,29 @@ func (t *compactionTrigger) reCalcSegmentMaxNumOfRows(collectionID UniqueID, isD
280278
281279// TODO: Update segment info should be written back to Etcd.
282280func (t * compactionTrigger ) updateSegmentMaxSize (segments []* SegmentInfo ) (bool , error ) {
283- ctx := context .Background ()
284-
285281 if len (segments ) == 0 {
286282 return false , nil
287283 }
288284
289285 collectionID := segments [0 ].GetCollectionID ()
290- resp , err := t .indexCoord .DescribeIndex (ctx , & indexpb.DescribeIndexRequest {
291- CollectionID : collectionID ,
292- IndexName : "" ,
293- })
294- if err != nil {
295- return false , err
296- }
286+ indexInfos := t .meta .GetIndexesForCollection (segments [0 ].GetCollectionID (), "" )
297287
298288 isDiskANN := false
299- for _ , indexInfo := range resp .IndexInfos {
300- indexParamsMap := funcutil .KeyValuePair2Map (indexInfo .IndexParams )
301- if indexType , ok := indexParamsMap ["index_type" ]; ok {
302- if indexType == indexparamcheck .IndexDISKANN {
303- // If index type is DiskANN, recalc segment max size here.
304- isDiskANN = true
305- newMaxRows , err := t .reCalcSegmentMaxNumOfRows (collectionID , true )
306- if err != nil {
307- return false , err
308- }
309- if len (segments ) > 0 && int64 (newMaxRows ) != segments [0 ].GetMaxRowNum () {
310- log .Info ("segment max rows recalculated for DiskANN collection" ,
311- zap .Int64 ("old max rows" , segments [0 ].GetMaxRowNum ()),
312- zap .Int64 ("new max rows" , int64 (newMaxRows )))
313- for _ , segment := range segments {
314- segment .MaxRowNum = int64 (newMaxRows )
315- }
289+ for _ , indexInfo := range indexInfos {
290+ indexType := getIndexType (indexInfo .IndexParams )
291+ if indexType == indexparamcheck .IndexDISKANN {
292+ // If index type is DiskANN, recalc segment max size here.
293+ isDiskANN = true
294+ newMaxRows , err := t .reCalcSegmentMaxNumOfRows (collectionID , true )
295+ if err != nil {
296+ return false , err
297+ }
298+ if len (segments ) > 0 && int64 (newMaxRows ) != segments [0 ].GetMaxRowNum () {
299+ log .Info ("segment max rows recalculated for DiskANN collection" ,
300+ zap .Int64 ("old max rows" , segments [0 ].GetMaxRowNum ()),
301+ zap .Int64 ("new max rows" , int64 (newMaxRows )))
302+ for _ , segment := range segments {
303+ segment .MaxRowNum = int64 (newMaxRows )
316304 }
317305 }
318306 }
@@ -365,7 +353,7 @@ func (t *compactionTrigger) handleGlobalSignal(signal *compactionSignal) {
365353 break
366354 }
367355
368- group .segments = FilterInIndexedSegments (t .handler , t .indexCoord , group .segments ... )
356+ // group.segments = FilterInIndexedSegments(t.handler, t.meta , group.segments...)
369357
370358 isDiskIndex , err := t .updateSegmentMaxSize (group .segments )
371359 if err != nil {
@@ -734,7 +722,7 @@ func reverseGreedySelect(candidates []*SegmentInfo, free int64, maxSegment int)
734722
735723func (t * compactionTrigger ) getCandidateSegments (channel string , partitionID UniqueID ) []* SegmentInfo {
736724 segments := t .meta .GetSegmentsByChannel (channel )
737- segments = FilterInIndexedSegments (t .handler , t .indexCoord , segments ... )
725+ segments = FilterInIndexedSegments (t .handler , t .meta , segments ... )
738726 var res []* SegmentInfo
739727 for _ , s := range segments {
740728 if ! isSegmentHealthy (s ) ||
0 commit comments