enhance: remove deprecated lazy load code (milvus-io#47590)

Related to milvus-io#44452

Remove the deprecated lazy load feature which has been superseded by
warmup-related parameters. This cleanup includes:

- Remove AddFieldDataInfoForSealed from C++ segcore layer
- Remove IsLazyLoad() method and isLazyLoad field from segment
- Remove lazy load checks in proxy alterCollectionTask
- Remove DiskCache lazy load handling in search/retrieve paths
- Remove LazyLoadEnableKey constant and related helper functions
- Update mock files to reflect interface changes

---------

Signed-off-by: Congqi Xia <congqi.xia@zilliz.com>

Author: congqixia

configs/milvus.yaml

@@ -531,13 +531,6 @@ queryNode:
     growingMmapEnabled: false
     fixedFileSizeForMmapAlloc: 1 # tmp file size for mmap chunk manager
     maxDiskUsagePercentageForMmapAlloc: 50 # disk percentage used in mmap chunk manager
-  lazyload:
-    enabled: false # Enable lazyload for loading data
-    waitTimeout: 30000 # max wait timeout duration in milliseconds before start to do lazyload search and retrieve
-    requestResourceTimeout: 5000 # max timeout in milliseconds for waiting request resource for lazy load, 5s by default
-    requestResourceRetryInterval: 2000 # retry interval in milliseconds for waiting request resource for lazy load, 2s by default
-    maxRetryTimes: 1 # max retry times for lazy load, 1 by default
-    maxEvictPerRetry: 1 # max evict count for lazy load, 1 by default
   indexOffsetCacheEnabled: false # enable index offset cache for some scalar indexes, now is just for bitmap index, enable this param can improve performance for retrieving raw data from index
   scheduler:
     receiveChanSize: 10240

internal/core/src/segcore/segment_c.cpp

@@ -794,26 +794,6 @@ DropSealedSegmentJSONIndex(CSegmentInterface c_segment,
     }
 }
 
-CStatus
-AddFieldDataInfoForSealed(CSegmentInterface c_segment,
-                          CLoadFieldDataInfo c_load_field_data_info) {
-    SCOPE_CGO_CALL_METRIC();
-
-    try {
-        auto segment_interface =
-            reinterpret_cast<milvus::segcore::SegmentInterface*>(c_segment);
-        auto segment =
-            dynamic_cast<milvus::segcore::SegmentSealed*>(segment_interface);
-        AssertInfo(segment != nullptr, "segment conversion failed");
-        auto load_info =
-            static_cast<LoadFieldDataInfo*>(c_load_field_data_info);
-        segment->AddFieldDataInfoForSealed(*load_info);
-        return milvus::SuccessCStatus();
-    } catch (std::exception& e) {
-        return milvus::FailureCStatus(milvus::UnexpectedError, e.what());
-    }
-}
-
 void
 RemoveFieldFile(CSegmentInterface c_segment, int64_t field_id) {
     SCOPE_CGO_CALL_METRIC();

internal/core/src/segcore/segment_c.h

@@ -240,10 +240,6 @@ DropSealedSegmentJSONIndex(CSegmentInterface c_segment,
                            int64_t field_id,
                            const char* nested_path);
 
-CStatus
-AddFieldDataInfoForSealed(CSegmentInterface c_segment,
-                          CLoadFieldDataInfo c_load_field_data_info);
-
 //////////////////////////////    interfaces for SegmentInterface    //////////////////////////////
 CStatus
 ExistPk(CSegmentInterface c_segment,

internal/mocks/util/mock_segcore/mock_CSegment.go

@@ -1343,18 +1343,9 @@ func hasWarmupProp(props ...*commonpb.KeyValuePair) bool {
 	return false
 }
 
-func hasLazyLoadProp(props ...*commonpb.KeyValuePair) bool {
-	for _, p := range props {
-		if p.GetKey() == common.LazyLoadEnableKey {
-			return true
-		}
-	}
-	return false
-}
-
 func hasPropInDeletekeys(keys []string) string {
 	for _, key := range keys {
-		if key == common.MmapEnabledKey || key == common.LazyLoadEnableKey || common.IsWarmupKey(key) {
+		if key == common.MmapEnabledKey || common.IsWarmupKey(key) {
 			return key
 		}
 	}
@@ -1405,16 +1396,15 @@ func (t *alterCollectionTask) PreExecute(ctx context.Context) error {
 
 	if len(t.GetProperties()) > 0 {
 		hasMmap := hasMmapProp(t.Properties...)
-		hasLazyLoad := hasLazyLoadProp(t.Properties...)
 		hasWarmup := hasWarmupProp(t.Properties...)
-		if hasMmap || hasLazyLoad || hasWarmup {
+		if hasMmap || hasWarmup {
 			loaded, err := isCollectionLoaded(ctx, t.mixCoord, t.CollectionID)
 			if err != nil {
 				return err
 			}
 			if loaded {
 				// keeping the original error msg here for compatibility
-				if hasMmap || hasLazyLoad {
+				if hasMmap {
 					return merr.WrapErrCollectionLoaded(t.CollectionName, "can not alter mmap properties if collection loaded")
 				}
 				if hasWarmup {

internal/proxy/task.go

@@ -6145,11 +6145,6 @@ func TestHasPropInDeletekeys(t *testing.T) {
 		assert.Equal(t, common.MmapEnabledKey, hasPropInDeletekeys(keys))
 	})
 
-	t.Run("has lazyload key", func(t *testing.T) {
-		keys := []string{common.LazyLoadEnableKey}
-		assert.Equal(t, common.LazyLoadEnableKey, hasPropInDeletekeys(keys))
-	})
-
 	t.Run("has generic warmup key", func(t *testing.T) {
 		keys := []string{common.WarmupKey}
 		assert.Equal(t, common.WarmupKey, hasPropInDeletekeys(keys))

internal/proxy/task_test.go

@@ -161,12 +161,6 @@ func (node *QueryNode) loadIndex(ctx context.Context, req *querypb.LoadSegmentsR
 			continue
 		}
 
-		if localSegment.IsLazyLoad() {
-			localSegment.SetLoadInfo(info)
-			localSegment.SetNeedUpdatedVersion(req.GetVersion())
-			node.manager.DiskCache.MarkItemNeedReload(ctx, localSegment.ID())
-			return nil
-		}
 		err := node.loader.LoadIndex(ctx, localSegment, info, req.Version)
 		if err != nil {
 			log.Warn("failed to load index", zap.Error(err))
@@ -200,12 +194,6 @@ func (node *QueryNode) loadStats(ctx context.Context, req *querypb.LoadSegmentsR
 			continue
 		}
 
-		if localSegment.IsLazyLoad() {
-			localSegment.SetLoadInfo(info)
-			localSegment.SetNeedUpdatedVersion(req.GetVersion())
-			node.manager.DiskCache.MarkItemNeedReload(ctx, localSegment.ID())
-			return nil
-		}
 		err := node.loader.LoadJSONIndex(ctx, localSegment, info)
 		if err != nil {
 			log.Warn("failed to load stats", zap.Error(err))

internal/querynodev2/handlers.go

@@ -31,16 +31,13 @@ import (
 
 	"go.uber.org/atomic"
 	"go.uber.org/zap"
-	"golang.org/x/sync/singleflight"
 
 	"github.com/milvus-io/milvus-proto/go-api/v2/commonpb"
-	"github.com/milvus-io/milvus/internal/querynodev2/segments/metricsutil"
 	"github.com/milvus-io/milvus/pkg/v2/eventlog"
 	"github.com/milvus-io/milvus/pkg/v2/log"
 	"github.com/milvus-io/milvus/pkg/v2/metrics"
 	"github.com/milvus-io/milvus/pkg/v2/proto/datapb"
 	"github.com/milvus-io/milvus/pkg/v2/proto/querypb"
-	"github.com/milvus-io/milvus/pkg/v2/util/cache"
 	"github.com/milvus-io/milvus/pkg/v2/util/lock"
 	"github.com/milvus-io/milvus/pkg/v2/util/merr"
 	"github.com/milvus-io/milvus/pkg/v2/util/metautil"
@@ -74,96 +71,16 @@ func IncreaseVersion(version int64) SegmentAction {
 type Manager struct {
 	Collection CollectionManager
 	Segment    SegmentManager
-	DiskCache  cache.Cache[int64, Segment]
 	Loader     Loader
 }
 
 func NewManager() *Manager {
-	diskCap := paramtable.Get().QueryNodeCfg.DiskCacheCapacityLimit.GetAsSize()
-
 	segMgr := NewSegmentManager()
-	sf := singleflight.Group{}
 	manager := &Manager{
 		Collection: NewCollectionManager(),
 		Segment:    segMgr,
 	}
 
-	manager.DiskCache = cache.NewCacheBuilder[int64, Segment]().WithLazyScavenger(func(key int64) int64 {
-		segment := segMgr.GetWithType(key, SegmentTypeSealed)
-		if segment == nil {
-			return 0
-		}
-		return int64(segment.ResourceUsageEstimate().DiskSize)
-	}, diskCap).WithLoader(func(ctx context.Context, key int64) (Segment, error) {
-		log := log.Ctx(ctx)
-		log.Debug("cache missed segment", zap.Int64("segmentID", key))
-		segment := segMgr.GetWithType(key, SegmentTypeSealed)
-		if segment == nil {
-			// the segment has been released, just ignore it
-			log.Warn("segment is not found when loading", zap.Int64("segmentID", key))
-			return nil, merr.ErrSegmentNotFound
-		}
-		info := segment.LoadInfo()
-		_, err, _ := sf.Do(fmt.Sprint(segment.ID()), func() (nop interface{}, err error) {
-			cacheLoadRecord := metricsutil.NewCacheLoadRecord(getSegmentMetricLabel(segment))
-			cacheLoadRecord.WithBytes(segment.ResourceUsageEstimate().DiskSize)
-			defer func() {
-				cacheLoadRecord.Finish(err)
-			}()
-
-			collection := manager.Collection.Get(segment.Collection())
-			if collection == nil {
-				return nil, merr.WrapErrCollectionNotLoaded(segment.Collection(), "failed to load segment fields")
-			}
-
-			err = manager.Loader.LoadLazySegment(ctx, segment, info)
-			return nil, err
-		})
-		if err != nil {
-			log.Warn("cache sealed segment failed", zap.Error(err))
-			return nil, err
-		}
-		return segment, nil
-	}).WithFinalizer(func(ctx context.Context, key int64, segment Segment) error {
-		log := log.Ctx(ctx)
-		log.Debug("evict segment from cache", zap.Int64("segmentID", key))
-		cacheEvictRecord := metricsutil.NewCacheEvictRecord(getSegmentMetricLabel(segment))
-		cacheEvictRecord.WithBytes(segment.ResourceUsageEstimate().DiskSize)
-		defer cacheEvictRecord.Finish(nil)
-		segment.Release(ctx, WithReleaseScope(ReleaseScopeData))
-		return nil
-	}).WithReloader(func(ctx context.Context, key int64) (Segment, error) {
-		log := log.Ctx(ctx)
-		segment := segMgr.GetWithType(key, SegmentTypeSealed)
-		if segment == nil {
-			// the segment has been released, just ignore it
-			log.Debug("segment is not found when reloading", zap.Int64("segmentID", key))
-			return nil, merr.ErrSegmentNotFound
-		}
-
-		localSegment := segment.(*LocalSegment)
-		err := manager.Loader.LoadIndex(ctx, localSegment, segment.LoadInfo(), segment.NeedUpdatedVersion())
-		if err != nil {
-			log.Warn("reload segment failed", zap.Int64("segmentID", key), zap.Error(err))
-			return nil, merr.ErrSegmentLoadFailed
-		}
-		if err := localSegment.RemoveUnusedFieldFiles(); err != nil {
-			log.Warn("remove unused field files failed", zap.Int64("segmentID", key), zap.Error(err))
-			return nil, merr.ErrSegmentReduplicate
-		}
-
-		return segment, nil
-	}).Build()
-
-	segMgr.registerReleaseCallback(func(s Segment) {
-		if s.Type() == SegmentTypeSealed {
-			// !!! We cannot use ctx of request to call Remove,
-			// Once context canceled, the segment will be leak in cache forever.
-			// Because it has been cleaned from segment manager.
-			manager.DiskCache.Remove(context.Background(), s.ID())
-		}
-	})
-
 	return manager
 }