User SDK
  │
  ▼
┌──────────┐    gRPC     ┌────────────┐   metadata query    ┌────────────┐
│  Proxy   │ ──────────▶ │ QueryCoord │ ◀────────────────── │ DataCoord  │
└──────────┘             └────────────┘                     └────────────┘
                               │                                 │
                               │ LoadSegments RPC                │ Provides Segment
                               ▼                                 │ Binlog/Index info
                         ┌────────────┐                          │
                         │ QueryNode  │ ◀─── download from ──────┘
                         │ (multiple) │      object storage
                         └────────────┘

┌─────────────────────────────────────────────────────────────────┐
│                      QueryCoord Perspective                     │
│                                                                 │
│  Next Target (desired state)    Current Target (confirmed state)│
│  "these segments should load"   "these segments confirmed ready"│
│  Source: pulled from DataCoord  Source: promoted from Next      │
│                                                                 │
│  ──────── Gap = segments that need loading ────────             │
│                                                                 │
│  Distribution (actual state)                                    │
│  "which segments are on which QueryNodes"                       │
│  Source: QueryNode heartbeat reports                            │
└─────────────────────────────────────────────────────────────────┘

The essence of all load behavior = detect the Gap between Target and Distribution,
then generate tasks to close it.

User calls LoadCollection(collectionName, replicaNumber=1, resourceGroups=["default"])
  │
  ▼
Proxy creates loadCollectionTask, enqueues in DDL task queue (ddQueue)
  │
  ▼
Task Scheduler executes in order: PreExecute() → Execute() → PostExecute()

// Pseudocode showing core logic
func (t *loadCollectionTask) Execute(ctx context.Context) error {
    // 1. Get collection schema
    schema := globalMetaCache.GetCollectionSchema(collectionName)

    // 2. Validate all vector fields have indexes (load requires indexes)
    ValidateIndexExistenceOnVectorFields(schema)

    // 3. Build fieldID → indexID mapping
    fieldIndexID := map[int64]int64{
        vectorFieldID: indexID,  // each vector field maps to its index
    }

    // 4. Build request and send to QueryCoord
    req := &querypb.LoadCollectionRequest{
        CollectionID:   collectionID,
        ReplicaNumber:  replicaNumber,  // default 1
        ResourceGroups: resourceGroups, // default ["default"]
        FieldIndexID:   fieldIndexID,
        LoadFields:     loadFields,     // list of fields to load
        Schema:         schema,
    }
    return queryCoord.LoadCollection(ctx, req)
}

QueryCoord.LoadCollection(req)
  │
  ├─ If req.Refresh=true → refreshCollection() path
  │
  ├─ Determine replica count and resource groups:
  │   ├─ User specified → use directly
  │   ├─ Not specified → query broker for collection/database-level pre-config
  │   └─ Neither → use cluster-level defaults:
  │       ├─ queryCoord.clusterLevelLoadReplicaNumber = 1
  │       └─ queryCoord.clusterLevelLoadResourceGroups = ["default"]
  │
  ├─ Conflict detection: if collection already loaded with different config → UpdateLoadConfigJob
  │
  └─ Create LoadCollectionJob → submit to jobScheduler → wait for completion

func (job *LoadCollectionJob) PreExecute() error {
    // Validate replicaNumber > 0 (default 1)
    // Validate resourceGroups non-empty (default ["default"])
    // If collection already loaded, check for config conflicts:
    //   - replicaNumber differs → error "can't change the replica number"
    //   - loadFields differ → error "can't change the load field list"
    //   - resourceGroups differ → error "can't change the resource groups"
}

Step 1: Get target Partitions
    broker.GetPartitions(collectionID) → get all partitionIDs from RootCoord
    Filter out already-loaded partitions
    │
Step 2: Clean up stale Replicas (for new collection)
    meta.ReplicaManager.RemoveCollection()
    │
Step 3: Get Collection info
    broker.DescribeCollection() → get schema, virtual channel names
    │
Step 4: Create Replicas ★
    utils.SpawnReplicasWithRG() → create replicas in resource group
    Each replica is assigned to specific QueryNode nodes
    │
Step 5: Create Partition metadata
    Each partition gets PartitionLoadInfo:
      Status: LoadStatus_Loading  ← initial state is "loading"
    │
Step 6: Create Collection metadata
    CollectionLoadInfo:
      Status: LoadStatus_Loading
      LoadType: LoadType_LoadCollection
      LoadFields, FieldIndexID, ReplicaNumber, etc.
    │
Step 7: Persist metadata
    meta.CollectionManager.PutCollection(collection, partitions...)
    │
Step 8: Update Next Target ★★★
    targetObserver.UpdateNextTarget(collectionID)
      → pulls segment and channel info from DataCoord
      → tells QueryCoord "which segments this collection needs to load"
    │
Step 9: Register with Collection Observer
    collectionObserver.LoadCollection(collectionID)
      → Observer starts periodically monitoring load progress

UpdateCollectionNextTarget(collectionID):
  │
  ├─ Call DataCoord.GetRecoveryInfoV2() (with 10 retries)
  │   Returns:
  │   ├─ VchannelInfo: channel info (checkpoint, dropped/flushed segment lists)
  │   └─ SegmentInfo: segment info (binlog paths, sizes, row counts, etc.)
  │
  ├─ Filter: only keep segments belonging to loaded partitions
  │
  ├─ Extract L0 (delta) segments
  │
  └─ Create CollectionTarget object, store in targetMgr.next

SegmentChecker (runs every 3s):
  │
  ├─ Iterate all Loading-state collections
  │
  ├─ For each collection:
  │   ├─ Get segment list from next target
  │   ├─ Compare with current state, find unloaded segments
  │   └─ Create SegmentTask (Grow action) for each unloaded segment
  │
  └─ Submit SegmentTasks to TaskScheduler
      │
      ├─ Scheduler sorts by priority:
      │   ├─ High: Channel tasks
      │   ├─ Normal: Segment tasks (load)
      │   └─ Low: Balance tasks
      │
      └─ Execute tasks: send LoadSegments RPC to target QueryNode

// Task structure
type SegmentTask struct {
    segmentID    int64
    actions      []Action    // a segment can have multiple actions
    priority     TaskPriority
}

// Action types
type Action struct {
    Node  int64      // target QueryNode
    Type  ActionType // Grow (load) / Reduce (unload) / Update (update)
}

type SegmentInfo struct {
    SegmentID   int64
    PartitionID int64
    NumRows     int64
    State       SegmentState  // Flushed, Sealed, Growing
    Binlogs     []*FieldBinlog  // insert data binlog paths
    Deltalogs   []*FieldBinlog  // delete data delta logs
    Statslogs   []*FieldBinlog  // statistics (bloom filters, etc.)
    IndexInfos  []*FieldIndexInfo // index file info
    MemorySize  int64           // estimated memory size
    Level       SegmentLevel    // L0 (delta) or regular
}

QueryNode.LoadSegments(req):
  │
  ├─ Validate node health
  ├─ Check if segment has index info
  ├─ If needTransfer=true → delegate to shard delegator
  └─ Otherwise:
      ├─ manager.Collection.PutOrRef(segmentID) → increment reference count
      └─ loader.Load(segments, loadScope) → trigger actual loading

func NewLoader() {
    ioPoolSize = min(max(CPU * 8, 32), 256)  // IO concurrency pool size
    // can be overridden via queryNode.ioPoolSize
}

segmentLoader.Load(segments):
  │
  Step 1: Filter binlog paths
  │  ├─ Keep only fields specified in loadFields
  │  └─ Skip already-loaded or currently-loading segments
  │
  Step 2: Resource acquisition ★★★
  │  requestResource(infos):
  │  ├─ Calculate estimated memory: LoadMemoryUsageFactor(default 2) × estimatedSize
  │  ├─ Check memory: physicalMemory + committed < totalMemory × threshold(90%)
  │  ├─ Check disk: diskUsage + committed < DiskCapacityLimit
  │  ├─ If LazyLoadEnabled=true → skip resource check
  │  └─ Return allowed concurrency level (ConcurrencyLevel)
  │
  Step 3: Create Segment objects
  │  NewSegment() → pre-allocate memory placeholder
  │
  Step 4: Parallel loading ★★★
  │  for each segment (concurrency = ConcurrencyLevel):
  │    └─ loadSegmentFunc(segment):
  │        │
  │        ├─ If L0 segment → load deltalogs only
  │        │
  │        └─ If regular sealed segment:
  │            │
  │            ├─ LoadSegment(): load index and raw data
  │            │   ├─ Load index files (from object storage)
  │            │   ├─ Decompress data
  │            │   └─ Populate bloom filter
  │            │
  │            ├─ loadDeltalogs(): load delete logs
  │            │
  │            ├─ loadBloomFilterSet(): load bloom filters
  │            │
  │            ├─ manager.Segment.Put(segment): register with manager
  │            │
  │            └─ notifyLoadFinish(): notify load complete
  │
  Step 5: Wait for all parallel tasks to complete
  │  waitSegmentLoadDone()
  │
  └─ Return list of loaded segments

LoadSegment(segment, loadInfo):
  │
  ├─ 1. Load vector index (most time-consuming)
  │     Download index files from object storage → deserialize → load into memory
  │     Supported indexes: HNSW, IVF_FLAT, IVF_SQ8, DiskANN, etc.
  │
  ├─ 2. Load scalar data
  │     Read per-field data from binlogs
  │
  ├─ 3. Load Delta Logs
  │     Read delete records from delta log
  │     Apply to segment's delete bitmap
  │
  └─ 4. Load Bloom Filter
      Read bloom filter from stats log
      Used to quickly determine if a PK exists in this segment

CollectionObserver (runs periodically, interval controlled by collectionObserverInterval):
  │
  for each LoadTask:
  │
  ├─ observeLoadStatus():
  │   │
  │   ├─ Get segment count from target
  │   │   targetNum = len(targetSegments) + channelTargetNum
  │   │
  │   ├─ Count loaded segments
  │   │   Query each QueryNode's load state via dist.LeaderViewManager
  │   │   Group by replica
  │   │
  │   ├─ Calculate load percentage ★
  │   │   loadPercentage = (loadedCount × 100) / (targetNum × replicaNum)
  │   │
  │   │   Example: 3 segments × 2 replicas = 6 targets
  │   │            4 loaded → (4 × 100) / 6 = 66%
  │   │
  │   └─ Update metadata
  │       meta.CollectionManager.UpdatePartitionLoadPercent(partID, percentage)
  │       When 100% is reached, triggers Proxy cache invalidation
  │
  └─ All partitions reach 100% → remove LoadTask → "Load task finish"

Next Target (desired state)
    ↓ all segments loaded on all replicas
Current Target (current state)
    → Collection becomes queryable

User calls GetLoadingProgress(collectionName)
  │
  ▼
Proxy → QueryCoord.ShowLoadCollections(collectionID)
  │
  ▼
QueryCoord calculates and returns:
  InMemoryPercentage: integer 0-100
  QueryServiceAvailable: bool (whether queries can start)

type UndoList struct {
    CollectionID      int64
    LackPartitions    []int64
    IsNewCollection   bool
    IsReplicaCreated  bool
    IsTargetUpdated   bool
}

func (u *UndoList) RollBack() {
    if u.IsNewCollection || u.IsReplicaCreated {
        meta.CollectionManager.RemoveCollection(u.CollectionID)
    } else {
        meta.CollectionManager.RemovePartition(u.CollectionID, u.LackPartitions...)
    }
    if u.IsTargetUpdated {
        targetObserver.ReleaseCollection(u.CollectionID)
    }
}

1. QueryNode-A receives LoadSegments RPC, calls requestResource()
2. Finds: physicalMemory(85%) + estimatedSize(10%) > threshold(90%)
3. Returns ErrOutOfMemory → Task marked as Failed
4. QueryCoord Scheduler retries:
   a. If other QueryNodes have sufficient memory → reassign to NodeB
   b. If all nodes are full → Task keeps failing, load progress stalls
   c. User needs to scale out QueryNodes or release other Collections

1. QueryNode calls LoadSegment() to download index files
2. Object storage (S3/MinIO) response times out
3. LoadSegment() returns error → Task marked as Failed
4. Scheduler detects segment still unloaded in next SegmentChecker round (3s later)
5. Recreates SegmentTask → retries LoadSegments RPC
6. If continuously failing, Load operation times out after LoadTimeoutSeconds (600s)

1. QueryNode-A crashes, all its segments disappear from Distribution
2. SegmentChecker detects: segment in CurrentTarget but not in Distribution
3. → Recovery scenario, generates Grow Task (LoadPriority_HIGH)
4. Two outcomes:
   a. NodeA restarts successfully → segments reload onto NodeA
   b. NodeA down for extended period → Balance Checker (Stopping Balance)
      migrates segments to other healthy nodes
5. Fully automatic, no manual intervention needed

// New parameter in 2.6
type loadCollectionTask struct {
    // ... same fields as 2.5
    priority commonpb.LoadPriority  // ← new: LoadPriority_High / LoadPriority_Low
}

2.5: Proxy → QueryCoord → LoadCollectionJob → direct metadata ops + direct RPC
2.6: Proxy → QueryCoord → broadcast AlterLoadConfigMessage to WAL → Job consumes WAL message

QueryCoord.LoadCollection(req):
  │
  ├─ Health check
  ├─ Refresh mode check
  │
  └─ broadcastAlterLoadConfigCollectionV2ForLoadCollection() ← new flow

broadcastAlterLoadConfigCollectionV2ForLoadCollection():
  │
  Step 1: Acquire Collection-level lock
  │  startBroadcastWithCollectionIDLock()
  │  → prevent concurrent load/release operations
  │
  Step 2: Get Collection metadata
  │  broker.DescribeCollection() → schema, partitionIDs
  │  broker.GetPartitions() → all partitionIDs
  │
  Step 3: Determine Replica configuration
  │  ├─ User specified → use directly
  │  └─ Not specified → getDefaultResourceGroupsAndReplicaNumber()
  │      → query pre-configured broker load info
  │
  Step 4: Validate resource group availability
  │  utils.AssignReplica() → check resource group has enough healthy nodes
  │
  Step 5: Build AlterLoadConfigRequest ★
  │  ├─ CurrentLoadConfig: current load state (read from meta)
  │  └─ ExpectedLoadConfig: target desired state
  │      ├─ ExpectedPartitionIDs
  │      ├─ ExpectedReplicaNumber (per resource group)
  │      ├─ ExpectedFieldIndexID
  │      ├─ ExpectedLoadFields
  │      └─ ExpectedPriority
  │
  Step 6: Generate and broadcast WAL message ★★★
      GenerateAlterLoadConfigMessage():
        ├─ Generate LoadFieldConfig array (fieldID + indexID)
        ├─ Generate LoadReplicaConfig array (per resource group)
        ├─ Compare previous and new header
        │   ├─ Same → return ErrIgnoredAlterLoadConfig (idempotent, no duplicate)
        │   └─ Different → continue
        └─ Broadcast AlterLoadConfigMessageBuilderV2 to streaming.WAL().ControlChannel()

LoadCollectionJob.Execute():
  │
  Step 1: Parse Replica configuration
  │  ├─ UseLocalReplicaConfig → use local cluster-level config
  │  └─ Otherwise → use replica config from message
  │
  Step 2: Create Replicas ★
  │  SpawnReplicasWithReplicaConfig():
  │  ├─ Create new replica ID (idempotent, reuse if exists)
  │  ├─ Assign to QueryNodes in resource group
  │  └─ Invalidate Proxy shard leader cache
  │
  Step 3: Persist Load metadata
  │  ├─ PartitionLoadInfo: Status = Loading
  │  └─ CollectionLoadInfo: Status = Loading, LoadType = LoadCollection
  │
  Step 4: Handle Partition release (if any)
  │  targetObserver.ReleasePartition()
  │
  Step 5: Update Next Target
  │  targetObserver.UpdateNextTarget(collectionID)
  │  → pull segment metadata from DataCoord
  │
  Step 6: Register with Collection Observer
  │  collectionObserver.LoadPartitions(ctx, collectionID, partitionIDs)
  │
  Step 7: Wait for release completion (if any)
  │  WaitCurrentTargetUpdated()
  │
  └─ Return

type SegmentTask struct {
    *baseTask
    segmentID    int64
    loadPriority commonpb.LoadPriority  // ← new in 2.6
    actions      []Action
}

// 2.5 Action types
Grow    // load
Reduce  // unload
Update  // update

// New Action types in 2.6
StatsUpdate  // update statistics
DropIndex    // drop index
Reopen       // reload (new config)

QueryNode.LoadSegments(req):
  │
  ├─ Validate node health
  ├─ Check index info
  │
  └─ Route based on LoadScope:
      ├─ DataScope_Delta → loadDeltaLogs()     (load delete logs only)
      ├─ DataScope_Index → loadIndex()         (load index only)
      ├─ DataScope_Stats → loadStats()         (load stats only) ← new in 2.6
      ├─ DataScope_All   → loader.Load()       (full load)
      └─ Reopen          → reopenSegments()    (reload) ← new in 2.6

segmentLoader.Load(segments):
  │
  Steps 1-3: Same as 2.5 (filter, resource acquisition, create Segment)
  │
  Step 4: Parallel loading (enhanced)
  │  for each segment (concurrency = ConcurrencyLevel):
  │    └─ LoadSegment(segment, loadInfo):
  │        │
  │        ├─ Load indexes ★ (Mmap enhanced)
  │        │   for each index (vector/scalar):
  │        │     ├─ enableMmap = isIndexMmapEnable(fieldSchema, indexInfo)
  │        │     │   ├─ Vector index: MmapVectorIndex (default false)
  │        │     │   └─ Scalar index: MmapScalarIndex (default false)
  │        │     │
  │        │     ├─ Determine warmup strategy (from field schema):
  │        │     │   ├─ no_warmup: no pre-load, on-demand on first access
  │        │     │   ├─ async_warmup: background async pre-load
  │        │     │   └─ sync_warmup: synchronous pre-load (blocks until done)
  │        │     │
  │        │     └─ segcore.LoadIndex():
  │        │         ├─ Mmap mode: file-mapped to virtual memory, OS manages page cache
  │        │         └─ Non-Mmap: fully loaded into memory
  │        │
  │        ├─ Load raw data ★ (Mmap enhanced)
  │        │   for each field:
  │        │     ├─ enableMmap = isDataMmapEnable(fieldSchema)
  │        │     │   ├─ Vector field: MmapVectorField (default true in 2.6)
  │        │     │   └─ Scalar field: MmapScalarField (default false)
  │        │     │
  │        │     └─ segcore.LoadFieldData(enableMmap):
  │        │         ├─ Mmap mode: write to disk cache → mmap → apply warmup
  │        │         └─ Non-Mmap: load directly into memory
  │        │
  │        ├─ Load Delta Logs
  │        │
  │        └─ Load Bloom Filter
  │
  └─ Return

Mmap loading mode:
  │
  ├─ Vector indexes (HNSW, IVF, DiskANN):
  │   MmapVectorIndex = false (default off)
  │   → index file written to disk cache
  │   → mmap() maps to virtual address space
  │   → actual memory managed by OS on demand
  │   → minimal memory usage (only page cache overhead)
  │
  ├─ Scalar indexes (BitmapIndex, etc.):
  │   MmapScalarIndex = false (default off)
  │   → same as above
  │
  ├─ Vector raw data:
  │   MmapVectorField = true (default ON in 2.6)
  │   → raw data mmap-mapped
  │   → suitable for large-scale vector storage
  │
  └─ Scalar raw data:
      MmapScalarField = false (default off)
      → loaded directly into memory (scalar data typically small, faster in-memory)
      → must be explicitly enabled

Warmup strategies (configured via field schema type_params):
  ├─ no_warmup:    fully on-demand; page faults on first query
  │                Pros: fastest load  Cons: high first-query latency
  │                Use for: cold data, infrequently queried fields
  │
  ├─ async_warmup: background async page loading; does not block LoadSegment return
  │                Pros: fast load + background warmup  Cons: queries may still be slow during warmup
  │                Use for: recommended for most scenarios
  │
  └─ sync_warmup:  synchronous load of all pages; blocks until complete
                   Pros: full speed immediately after load  Cons: slowest load
                   Use for: hot data with strict query latency requirements

Lazy Load enable conditions:
  collection.Properties sets "lazy_load_enabled" = true
  AND segmentType != SegmentTypeGrowing

Lazy Load flow:
  │
  ├─ Skip resource check during Load()
  │   → segment immediately marked as "loaded" (data not actually loaded)
  │
  ├─ First query triggers actual loading:
  │   lazyLoadRequestResourceWithRetry():
  │     ├─ Loop requesting resources, 2s interval (lazyload.requestResourceRetryInterval)
  │     ├─ 5s timeout (lazyload.requestResourceTimeout)
  │     ├─ Max 1 retry (lazyload.maxRetryTimes)
  │     ├─ On OOM: try evicting rarely-used segments (LazyLoadMaxEvictPerRetry)
  │     └─ Resources available → execute actual loading
  │
  └─ Advantages:
      ├─ LoadCollection returns near-instantly
      ├─ Suitable for large collections without need to query all data simultaneously
      └─ Cold data automatically evicted, hot data retained in memory

                    ┌─────────────────────────────────────┐
                    │     All paths converge at:           │
                    │  cluster.LoadSegments(node, req)     │
                    │  (executor.go, line ~308)            │
                    └──────────────┬──────────────────────┘
                                   │
               ┌───────────────────┼───────────────────┐
               │                   │                   │
      ┌────────┴────────┐  ┌───────┴──────┐  ┌────────┴────────┐
      │  3 Checkers     │  │  1 Observer  │  │  1 Checker      │
      │ (produce tasks) │  │ (updates     │  │  (indirect dep) │
      └────────┬────────┘  │  Target)     │  └────────┬────────┘
               │           └───────┬──────┘           │
   ┌───────────┼──────────┐        │                  │
   │           │          │        │                  │
   ▼           ▼          ▼        ▼                  ▼
SegmentChecker IndexChecker BalanceChecker TargetObserver ChannelChecker
(direct Grow)  (Update/     (Grow+Reduce  (pull new      (prerequisite:
               Stats tasks)  migration)    Target →       channel must be
                                          drives          in place before
                                          Checkers)       segment loads)
   │
   ├─ Scenario 1: Initial Load (user calls LoadCollection)
   ├─ Scenario 2: Recovery (node failure, reload) → HIGH priority
   ├─ Scenario 3: Handoff (Growing→Sealed transition) → LOW priority
   └─ Scenario 4: Refresh/Import (load after data import)

[Trigger path 1-9] → detect Gap between Target and Distribution
  ↓
Generate Task (Grow / Reduce / Reopen / StatsUpdate etc.)
  ↓
Scheduler.Add(task) → queue by priority
  ↓
Executor.ExecuteTask()
  ↓
Executor.executeSegmentAction()
  ↓
Executor.loadSegment()  [Grow/Reopen/StatsUpdate all go here]
  ↓
getMetaInfo() → getLoadInfo() → packLoadSegmentRequest()
  ↓
ChannelDistManager.GetShardLeader() → determine target node
  ↓
cluster.LoadSegments(ctx, nodeID, request)  ← final RPC call

SegmentChecker.Check():
  │
  for each loaded collection:
  │
  ├─ checkReplica(replica):
  │   │
  │   └─ getSealedSegmentDiff():
  │       │
  │       ├─ Iterate segments in NextTarget
  │       │   Compare with SegmentDistribution (loaded segments)
  │       │
  │       ├─ Find "should have but doesn't" segments → need Grow
  │       │
  │       └─ Assign priority by scenario: ★★★ critical logic
  │           │
  │           ├─ Segment in CurrentTarget but not in Distribution
  │           │   → Recovery scenario → LoadPriority_HIGH
  │           │
  │           ├─ Collection not fully refreshed (Import scenario)
  │           │   → Refresh scenario → replica.LoadPriority() (user config)
  │           │
  │           └─ Segment in NextTarget but not in CurrentTarget
  │               → Handoff scenario → LoadPriority_LOW
  │
  └─ createSegmentLoadTasks():
      → group by shard
      → get shard leader
      → create SegmentTask (Grow action)
      → submit to Scheduler

if currentTargetExist {
    _, existOnCurrent := currentTargetMap[segment.GetID()]
    if existOnCurrent {
        // In CurrentTarget but not in Distribution → Recovery (node died)
        loadPriorities = append(loadPriorities, LoadPriority_HIGH)
    } else if collection != nil && !collection.IsRefreshed() {
        // Collection in refresh state (Import) → user-configured priority
        loadPriorities = append(loadPriorities, replica.LoadPriority())
    } else {
        // In NextTarget but not in CurrentTarget → Handoff (flush produced new segment)
        loadPriorities = append(loadPriorities, LoadPriority_LOW)
    }
}

IndexChecker.Check():
  │
  for each loaded collection:
  │
  ├─ Get all index definitions from broker
  │
  ├─ for each replica:
  │   for each segment in distribution:
  │     │
  │     ├─ checkSegment(): compare current IndexInfo with latest index from broker
  │     │
  │     ├─ If index has updates (new index build completed):
  │     │   → add to segmentsToUpdate list
  │     │
  │     └─ If JSON stats need update (EnabledJSONKeyStats=true):
  │         → add to segmentsToStats list
  │
  ├─ broker.GetIndexInfo(segmentIDs) → batch fetch latest index info (1024 per batch)
  │
  ├─ Generate Update Task (ActionTypeUpdate):
  │   → priority: TaskPriorityLow
  │   → reason: "missing index"
  │   → effect: QueryNode updates segment with new index
  │
  └─ Generate StatsUpdate Task (ActionTypeStatsUpdate): [if enabled]
      → priority: TaskPriorityLow
      → reason: "missing json stats"

BalanceChecker.Check():
  │
  ├─ Phase 1: Stopping Balance ★ High priority
  │   │
  │   ├─ Trigger: a node is marked RO (read-only/stopping) or ROSQ
  │   │   Detection: replica.RONodesCount() + ROSQNodesCount() > 0
  │   │
  │   ├─ Execute:
  │   │   getReplicaForStoppingBalance() → find replicas with RO nodes
  │   │   StoppingBalancer.BalanceReplica() → generate migration plan
  │   │
  │   ├─ Generate:
  │   │   ├─ Grow Task (load segment on target node): LoadPriority_HIGH
  │   │   └─ Reduce Task (unload on source node): executes after Grow completes
  │   │
  │   └─ Use: data migration before node goes offline, ensures no data loss
  │
  └─ Phase 2: Normal Balance → Low priority
      │
      ├─ Trigger conditions:
      │   ├─ AutoBalance = true (default on)
      │   ├─ Time since last balance exceeds AutoBalanceInterval (default 5s)
      │   └─ Phase 1 produced no tasks
      │
      ├─ Execute:
      │   ScoreBasedBalancer.BalanceReplica() → multi-factor scoring → migration plan
      │
      ├─ Scoring factors (2.6):
      │   ├─ RowCountFactor (0.4): row count weight
      │   ├─ SegmentCountFactor (0.4): segment count weight
      │   ├─ GlobalRowCountFactor (0.1): global row count weight
      │   └─ GlobalSegmentCountFactor (0.1): global segment count weight
      │
      ├─ Balance trigger threshold:
      │   ScoreUnbalanceTolerationFactor = 0.05
      │   → only triggers when imbalance between nodes > 5%
      │
      ├─ Generate:
      │   ├─ Grow Task (target node): LoadPriority_LOW
      │   └─ Reduce Task (source node): executes after Grow completes
      │
      └─ Batch limits:
          ├─ BalanceSegmentBatchSize = 5 (max 5 segments migrated per round)
          └─ BalanceChannelBatchSize = 1 (max 1 channel migrated per round)

Migrating a segment from NodeA → NodeB:

1. Create Grow Action: load segment on NodeB
   → send LoadSegments RPC to NodeB

2. Wait for Grow to complete

3. Create Reduce Action: unload segment on NodeA
   → send ReleaseSegments RPC to NodeA

4. Update Leader View: shard leader routes to NodeB

TargetObserver (runs continuously):
  │
  ├─ shouldUpdateNextTarget():
  │   ├─ Time since last update exceeds UpdateNextTargetInterval
  │   ├─ Or NextTargetSurviveTime expired
  │   └─ Or manually triggered externally
  │
  ├─ UpdateNextTarget(collectionID):
  │   │
  │   ├─ targetMgr.UpdateCollectionNextTarget():
  │   │   ├─ broker.GetRecoveryInfoV2() → get latest segment list from DataCoord
  │   │   │   Returns:
  │   │   │   ├─ New compacted segments (merged large segments)
  │   │   │   ├─ New flushed sealed segments
  │   │   │   └─ Deleted old segments
  │   │   │
  │   │   └─ Create new NextTarget (with latest segment list)
  │   │
  │   └─ Wait for segment data ready (if UpdateTargetNeedSegmentDataReady=true)
  │
  ├─ shouldUpdateCurrentTarget():
  │   ├─ All segments in NextTarget loaded on all replicas
  │   └─ Or timed out
  │
  └─ updateCurrentTarget():
      ├─ Promote NextTarget to CurrentTarget
      └─ Clear NextTarget
          │
          ▼
      SegmentChecker detects new segments in new CurrentTarget
      → automatically generates Grow Task → LoadSegments RPC

Before compaction:
  CurrentTarget: [seg1, seg2, seg3]    ← 3 small segments
  Distribution:  [seg1@Node1, seg2@Node1, seg3@Node2]

DataCoord executes compaction: seg1 + seg2 → seg4

After compaction:
  NextTarget:    [seg3, seg4]          ← seg4 is newly merged
  CurrentTarget: [seg1, seg2, seg3]    ← old ones still there

SegmentChecker detects:
  seg4 in NextTarget but not in Distribution → Grow (Handoff, LOW priority)

After seg4 loads:
  NextTarget promoted to CurrentTarget
  seg1, seg2 not in new CurrentTarget → Reduce (unload old segments)

Handoff complete flow:

1. DataNode flushes growing segment → produces sealed segment
   │
2. DataCoord updates segment list:
   │  ├─ Add sealed segment (seg_new)
   │  └─ Remove growing segment (seg_old)
   │
3. TargetObserver pulls update → NextTarget contains seg_new, not seg_old
   │
4. SegmentChecker.getSealedSegmentDiff():
   │  seg_new in NextTarget but not in Distribution
   │  and not in CurrentTarget → Handoff scenario
   │  → Grow Task: LoadPriority_LOW
   │
5. QueryNode loads seg_new (sealed segment)
   │
6. SegmentChecker.getGrowingSegmentDiff():
   │  seg_old (growing) in LeaderView but not in Target
   │  → Reduce Task: release growing segment
   │  → DataScope_Streaming
   │
7. Growing segment released, Sealed segment ready
   │
8. NextTarget promoted to CurrentTarget → handoff complete

Node crash and recovery flow:

1. QueryNode-A crashes
   │
2. ResourceObserver detects node offline
   │  → Node removed from replica's RW node list
   │  → Node's segments disappear from Distribution
   │
3. SegmentChecker.Check():
   │  segment in CurrentTarget but not in Distribution
   │  → Recovery scenario
   │  → Grow Task: LoadPriority_HIGH ← highest priority!
   │
4. Two recovery paths:
   │
   ├─ Path A: Node restarts successfully
   │   → segments reload onto restarted node
   │
   └─ Path B: Node down for extended period
       → segments scheduled to other healthy nodes
       (if stopping balance is active, Balance Checker handles this)

ChannelChecker.Check():
  │
  ├─ getDmChannelDiff():
  │   Compare channels in NextTarget vs ChannelDistribution
  │
  ├─ Missing channel:
  │   → Channel Load Task (ActionTypeGrow)
  │   → Priority: TaskPriorityHigh (all channel tasks are high priority)
  │   → Assignment strategy: RoundRobin
  │
  ├─ Excess channel:
  │   → Channel Reduce Task
  │   → Clean up released collection's channels
  │
  └─ Duplicate channel (same channel on multiple nodes):
      → Keep newer version, release old one

LeaderChecker.Check():
  │
  ├─ findNeedLoadedSegments():
  │   │
  │   ├─ Compare LeaderView (routing table) vs Distribution (actual distribution)
  │   │
  │   ├─ Segment in Distribution but missing in routing table:
  │   │   → LeaderSegmentTask (Grow): "add segment to leader view"
  │   │   → Update Delegator routing table
  │   │
  │   └─ Segment in routing table but pointing to wrong node:
  │       → Update route to correct node
  │
  └─ findNeedRemovedSegments():
      Segment in routing table but no longer in Distribution:
      → LeaderSegmentTask (Reduce): "remove segment from leader view"

Import/BulkInsert flow:
  │
  1. User executes BulkInsert → DataCoord creates new segments
  │
  2. User calls LoadCollection(refresh=true)
  │   or system auto-detects new segments
  │
  3. Collection marked as "not refreshed"
  │   collection.IsRefreshed() = false
  │
  4. TargetObserver updates NextTarget (includes newly imported segments)
  │
  5. SegmentChecker detects new segments:
  │   collection.IsRefreshed() == false → Refresh scenario
  │   → Grow Task: replica.LoadPriority() (user-configured, default HIGH)
  │
  6. All new segments loaded
  │   → collection.SetRefreshed(true)
  │   → Refresh complete

TaskPriorityHigh  ──── Channel tasks (all channel operations)
  │                    Stopping Balance segment tasks
  │
TaskPriorityNormal ─── User Load segment tasks
  │                    Growing segment release tasks
  │
TaskPriorityLow  ───── Index update (Update) tasks
                       Normal Balance segment tasks
                       Leader route update tasks
                       Handoff cleanup tasks
                       Stats update tasks

LoadPriority_HIGH  ──── Node failure recovery (data may be lost, most urgent)
  │                     Stopping Balance (node about to go offline)
  │                     User-initiated Load (default HIGH)
  │                     Import/Refresh (replica default HIGH)
  │
LoadPriority_LOW  ───── Handoff (Growing→Sealed, non-urgent)
                        Normal Balance (performance optimization, least urgent)

=== Milvus 2.5 ===

SDK → Proxy → QueryCoord.LoadCollection()
                │
                ▼
          LoadCollectionJob (direct execution)
                │
                ├─ Create Replica
                ├─ Persist metadata (Status: Loading)
                ├─ UpdateNextTarget (pull DataCoord segment info)
                └─ Register CollectionObserver
                      │
                      ▼
                SegmentChecker (every 3s)
                → Create SegmentTask (Grow)
                → Scheduler sends LoadSegments RPC to QueryNode
                      │
                      ▼
                QueryNode.LoadSegments()
                → Resource check → parallel load (index+data+delta+bloom)
                      │
                      ▼
                CollectionObserver detects 100% → Loaded


=== Milvus 2.6 ===

SDK → Proxy → QueryCoord.LoadCollection()
                │
                ▼
          broadcastAlterLoadConfigMessage() ← new step
                │
                ├─ Acquire lock
                ├─ Get metadata
                ├─ Validate resource groups
                ├─ Build AlterLoadConfigRequest
                └─ Broadcast to WAL ControlChannel ← key change
                      │
                      ▼
          WAL consumed → LoadCollectionJob (executed)
                │
                ├─ Create Replica (idempotent)
                ├─ Persist metadata (Status: Loading)
                ├─ UpdateNextTarget
                └─ Register CollectionObserver
                      │
                      ▼
                SegmentChecker (every 3s)
                → Create SegmentTask (Grow, with LoadPriority)
                → Scheduler sends LoadSegments RPC
                      │
                      ▼
                QueryNode.LoadSegments()
                → Resource check → parallel load
                → **Fine-grained Mmap control** (vector/scalar independent config)
                → **Warmup strategy** (no/async/sync)
                → **Enhanced Lazy Load** (retry + eviction)
                      │
                      ▼
                CollectionObserver detects 100% → Loaded

CheckerController.Start()
  │
  ├─ goroutine: ChannelChecker  (every 3s)   ─┐
  ├─ goroutine: SegmentChecker  (every 3s)    │
  ├─ goroutine: BalanceChecker  (every 300ms) ├─→ run in parallel, non-blocking
  ├─ goroutine: IndexChecker    (every 10s)   │
  └─ goroutine: LeaderChecker   (every 1s)   ─┘
                    │
                    ▼
              Each tick:
              Check(ctx) → generate Tasks → scheduler.Add(task)

  ChannelChecker ──────────▶ SegmentChecker ──────────▶ LeaderChecker
  "is channel in place?"      "is segment in place?"      "is routing correct?"
       │                          │
       │                          │
       ▼                          ▼
  BalanceChecker ◀──────────────────
  "is load balanced?"
       │
       ▼
  IndexChecker (independent)
  "is index up to date?"

Implicit execution ordering:
  1. Channel must be assigned first → otherwise SegmentChecker can't find shard leader, skips
  2. Segment must be loaded first → otherwise LeaderChecker can't update routing
  3. Balance is based on current distribution → needs Segment/Channel distribution info
  4. Index runs independently → doesn't depend on other Checkers

Scheduler
  ├─ waitQueue:    tasks waiting to execute (just submitted)
  ├─ processQueue: tasks being processed
  │
  └─ bucketed by priority:
      ├─ High bucket:   ChannelChecker tasks
      ├─ Normal bucket: SegmentChecker tasks
      └─ Low bucket:    BalanceChecker / IndexChecker / LeaderChecker tasks

scheduler.Add(newTask):
  │
  ├─ Check if existing task with same (replicaID, segmentID)
  │
  ├─ Exists and new task has higher priority:
  │   → Cancel old task, replace with new task
  │   → Example: Balance(Low) replaced by Recovery(Normal)
  │
  ├─ Exists and new task has same or lower priority:
  │   → Reject new task
  │   → Next Checker round will regenerate
  │
  └─ No conflict:
      → Enqueue normally

Per-QueryNode task concurrency limit:

  Total limit = ceil(CPUNum × QueryNodeTaskParallelismFactor)
    │
    ├─ Channel task limit = ceil(total × ChannelTaskCapFraction)
    │   Default ChannelTaskCapFraction = 0.3 (30%)
    │   Minimum = 1 (prevents starvation)
    │
    └─ Non-channel task limit = total - channel limit
        Minimum = 1 (prevents starvation)

  Both task types counted independently, no mutual preemption

Executor.Execute(task, step):
  │
  ├─ Check concurrency limit (channelTaskNum / nonChannelTaskNum)
  │   Over limit → reject, wait for next scheduling round
  │
  ├─ Execute asynchronously (spawn goroutine):
  │   │
  │   ├─ Segment-type Action:
  │   │   ├─ Grow / Update / StatsUpdate → loadSegment()
  │   │   │   ├─ getMetaInfo()        → get collection/partition metadata
  │   │   │   ├─ getLoadInfo()        → get segment binlog/index info
  │   │   │   ├─ packLoadSegmentRequest() → build request
  │   │   │   └─ cluster.LoadSegments(nodeID, req) → send RPC ★
  │   │   │
  │   │   └─ Reduce → releaseSegment()
  │   │       └─ cluster.ReleaseSegments(nodeID, req)
  │   │
  │   ├─ Channel-type Action:
  │   │   ├─ Grow → subscribeChannel()
  │   │   └─ Reduce → unsubscribeChannel()
  │   │
  │   └─ Leader-type Action:
  │       ├─ Grow → setDistribution() (update routing table)
  │       └─ Reduce → removeDistribution()
  │
  └─ Execution complete:
      ├─ Update Distribution Manager
      ├─ Set executedFlag (triggers distHandler to immediately pull)
      └─ Remove task

QueryCoord                                QueryNode
   │                                         │
   │  Every 500ms (DistPullInterval)          │
   ├─ GetDataDistribution(LastUpdateTs) ────▶│
   │                                         │
   │                                         ├─ Collect current state:
   │                                         │  ├─ Loaded sealed segments
   │                                         │  ├─ Subscribed channels
   │                                         ���  ├─ Shard leader views
   │                                         │  └─ Memory/CPU info
   │                                         │
   │◀──── GetDataDistributionResponse ───────┤
   │  (segments, channels, leaderViews,      │
   │   lastModifyTs, memCapacity, cpuNum)    │
   │                                         │
   ├─ Update heartbeat timestamp             │
   ├─ If lastModifyTs unchanged → skip       │
   └─ If changed → update Distribution Mgr  │

DistributionManager
  ├─ SegmentDistManager   → which nodes each segment is on
  │   internal/querycoordv2/meta/segment_dist_manager.go
  │
  ├─ ChannelDistManager   → which node each channel is on + LeaderView
  │   internal/querycoordv2/meta/channel_dist_manager.go
  │
  └─ Used by:
      ├─ SegmentChecker  → compare Target vs Distribution
      ├─ BalanceChecker  → analyze inter-node load differences
      ├─ LeaderChecker   → compare LeaderView vs Distribution
      └─ CollectionObserver → calculate load progress percentage

1. distHandler records lastUpdateTs (timestamp of last processing)
2. QueryNode returns LastModifyTs (timestamp of most recent state change)
3. If LastModifyTs <= lastUpdateTs → skip (no changes)
4. Otherwise → execute full Distribution update

On each successful GetDataDistribution:
  → NodeInfo.LastHeartbeat = now()
  → Update metric: QueryCoordLastHeartbeatTimeStamp

Heartbeat lag warning:
  if now() - LastHeartbeat > HeartBeatWarningLag (default 5s):
      → Log warning "QueryNode heartbeat lag too large"

Node failure determination:
  Determined by ResourceObserver / NodeManager based on heartbeat timeout
  → Triggers SegmentChecker Recovery and BalanceChecker Stopping Balance

User calls ReleaseCollection(collectionName)
  │
  ▼
┌──────────────────────────────────────────────────┐
│ Proxy                                            │
│ File: internal/proxy/impl.go                     │
│                                                  │
│ 1. Create releaseCollectionTask                  │
│ 2. Enqueue in DDL task queue                     │
│ 3. Get collectionID                              │
│ 4. Call QueryCoord.ReleaseCollection()           │
└──────────────┬───────────────────────────────────┘
               │
               ▼
┌──────────────────────────────────────────────────┐
│ QueryCoord                                       │
│ File: internal/querycoordv2/services.go          │
│                                                  │
│ 1. Broadcast DropLoadConfigMessageV2 to WAL (2.6)│
│                                                  │
│ 2. ReleaseCollectionJob.Execute():               │
│    ├─ Delete Collection metadata                 │
│    │   meta.CollectionManager.RemoveCollection() │
│    │                                             │
│    ├─ Delete all Replicas                        │
│    │   meta.ReplicaManager.RemoveCollection()    │
│    │                                             │
│    ├─ Clean up Target                            │
│    │   targetObserver.ReleaseCollection()        │
│    │   → targetMgr.RemoveCollection()            │
│    │                                             │
│    ├─ Invalidate Proxy caches                    │
│    │   proxyManager.InvalidateCollectionMetaCache│
│    │   proxyManager.InvalidateShardLeaderCache() │
│    │                                             │
│    └─ Wait for Release completion                │
│       WaitCollectionReleased()                   │
│       → Poll Distribution until all              │
│         segments/channels gone from all nodes    │
└──────────────┬───────────────────────────────────┘
               │
               ▼
┌──────────────────────────────────────────────────┐
│ QueryNode (passive cleanup)                      │
│ File: internal/querynodev2/services.go           │
│                                                  │
│ ReleaseSegments(segmentIDs):                     │
│   ├─ manager.Segment.Remove(segmentID)           │
│   │   → release memory / mmap mapping            │
│   └─ manager.Collection.Unref()                  │
│       → clean up collection metadata when        │
│         reference count drops to zero            │
└──────────────────────────────────────────────────┘

ReleasePartitions(partitionIDs):
  │
  ├─ Case 1: partitions still online after release
  │   → Broadcast AlterLoadConfigMessageV2 (updated partition list)
  │   → Only clean up segments for specified partitions
  │   → Collection remains queryable (other partitions)
  │
  └─ Case 2: all partitions offline after release
      → Broadcast DropLoadConfigMessageV2
      → Equivalent to ReleaseCollection
      → Clean up all resources