flowchart TD
    RC[RootCoord] -->|DDL事件| DC[DataCoord]
    P[Proxy] -->|AssignSegmentID| DC
    DN1[DataNode1] -.->|Flush完成| DC
    DN2[DataNode2] -.->|Compaction进度| DC
    
    DC -->|Flush任务| DN1
    DC -->|Compaction任务| DN2
    DC -->|Handoff通知| QC[QueryCoord]
    
    DC <-->|元数据读写| ETCD[etcd/TiKV]
    DC -->|清理文件| OS[Object Storage]
    
    DC -->|Metrics| Prom[Prometheus]
    
    style DC fill:#f9f,stroke:#333,stroke-width:4px

stateDiagram-v2
    [*] --> Created: CreateServer()
    Created --> Initializing: Init()
    Initializing --> Initializing: 初始化元数据/连接
    Initializing --> Healthy: Start()
    Healthy --> Healthy: 处理请求
    Healthy --> Stopping: Stop()
    Stopping --> [*]: 清理资源
    
    note right of Initializing
        1. 连接etcd
        2. 加载元数据
        3. 初始化SegmentManager
        4. 启动Compaction
        5. 启动GC
    end note
    
    note right of Healthy
        1. 接收Segment分配请求
        2. 调度Flush任务
        3. 调度Compaction任务
        4. 处理DataNode心跳
        5. 执行垃圾回收
    end note

flowchart TB
    subgraph Upstream["上游组件"]
        Proxy[Proxy<br/>数据写入入口]
        RC[RootCoord<br/>DDL协调器]
        QC[QueryCoord<br/>查询协调器]
    end
    
    subgraph DataCoord["DataCoord Server"]
        subgraph APILayer["API 层"]
            GRPC[gRPC Server<br/>对外接口]
        end
        
        subgraph CoreManagers["核心管理器"]
            direction TB
            SM[SegmentManager<br/>━━━━━━━━━━━━<br/>• Segment分配<br/>• 状态管理<br/>• Seal策略]
            
            CM[ChannelManager<br/>━━━━━━━━━━━━<br/>• Channel分配<br/>• 负载均衡<br/>• 故障转移]
            
            Meta[Meta<br/>━━━━━━━━━━━━<br/>• 内存缓存<br/>• 元数据持久化<br/>• 事务支持]
        end
        
        subgraph TaskSchedulers["任务调度层"]
            direction LR
            CTM[CompactionTriggerManager<br/>━━━━━━━━━━━━<br/>• L0 Compaction<br/>• Mix Compaction<br/>• Clustering]
            
            CI[CompactionInspector<br/>━━━━━━━━━━━━<br/>• 任务调度<br/>• 进度监控<br/>• 结果处理]
            
            II[IndexInspector<br/>━━━━━━━━━━━━<br/>• 索引任务调度<br/>• 构建监控]
        end
        
        subgraph BackgroundServices["后台服务"]
            GC[GarbageCollector<br/>━━━━━━━━━━━━<br/>• Dropped Segment清理<br/>• Binlog文件回收<br/>• 索引文件清理]
            
            Handler[Handler<br/>━━━━━━━━━━━━<br/>• Collection信息缓存<br/>• RootCoord交互<br/>• QueryCoord交互]
        end
        
        GRPC --> SM
        GRPC --> CM
        GRPC --> Meta
        GRPC --> CTM
        
        SM --> Meta
        CM --> Meta
        CTM --> Meta
        CI --> Meta
        II --> Meta
        GC --> Meta
        
        Handler --> Meta
    end
    
    subgraph Downstream["下游组件"]
        DN1[DataNode-1<br/>━━━━━━<br/>数据写入]
        DN2[DataNode-2<br/>━━━━━━<br/>数据写入]
        DNn[DataNode-N<br/>━━━━━━<br/>数据写入]
        
        Cluster[Cluster Manager<br/>━━━━━━━━━━━━<br/>DataNode通信层]
    end
    
    subgraph Storage["存储层"]
        direction TB
        ETCD[(etcd/TiKV<br/>━━━━━━━━<br/>元数据存储)]
        S3[(Object Storage<br/>━━━━━━━━<br/>MinIO/S3<br/>Binlog文件)]
    end
    
    %% 上游到 DataCoord
    Proxy -->|AssignSegmentID<br/>获取Segment分配| GRPC
    RC -->|WatchChannels<br/>创建Collection| GRPC
    RC -->|DropCollection<br/>删除Collection| GRPC
    QC -->|GetSegmentInfo<br/>查询Segment| GRPC
    
    %% DataCoord 到下游
    CM -->|Watch/Release Channel<br/>分配/释放| Cluster
    CI -->|Compaction Task<br/>压缩任务| Cluster
    SM -->|Flush Command<br/>刷盘命令| Cluster
    
    Cluster --> DN1
    Cluster --> DN2
    Cluster --> DNn
    
    %% DataCoord 到存储
    Meta <-->|Save/Load<br/>Segment Meta| ETCD
    CM <-->|Save/Load<br/>Channel Meta| ETCD
    GC -->|Delete<br/>Binlog Files| S3
    
    %% DataNode 反馈
    DN1 -.->|FlushCompleted<br/>CompactionCompleted| GRPC
    DN2 -.->|FlushCompleted<br/>CompactionCompleted| GRPC
    DNn -.->|FlushCompleted<br/>CompactionCompleted| GRPC
    
    style DataCoord fill:#e1f5ff,stroke:#1976d2,stroke-width:3px
    style CoreManagers fill:#fff4e6,stroke:#f57c00,stroke-width:2px
    style TaskSchedulers fill:#f3e5f5,stroke:#7b1fa2,stroke-width:2px
    style BackgroundServices fill:#e8f5e9,stroke:#388e3c,stroke-width:2px
    style Upstream fill:#ffebee,stroke:#c62828,stroke-width:2px
    style Downstream fill:#e0f2f1,stroke:#00695c,stroke-width:2px
    style Storage fill:#fce4ec,stroke:#880e4f,stroke-width:2px

flowchart LR
    subgraph Write["写入流程"]
        W1[Proxy请求分配] --> W2[SegmentManager分配]
        W2 --> W3[Meta持久化]
        W3 --> W4[返回SegmentID]
        W4 --> W5[DataNode写入数据]
        W5 --> W6[Segment达到阈值]
    end
    
    subgraph Flush["Flush流程"]
        F1[SegmentManager检测] --> F2[Seal Segment]
        F2 --> F3[通知DataNode Flush]
        F3 --> F4[DataNode写Binlog]
        F4 --> F5[更新Meta状态]
        F5 --> F6[通知QueryCoord]
    end
    
    subgraph Compact["Compaction流程"]
        C1[CompactionTrigger扫描] --> C2[计算收益]
        C2 --> C3[提交CompactionInspector]
        C3 --> C4[DataNode合并]
        C4 --> C5[写新Segment]
        C5 --> C6[Drop旧Segment]
    end
    
    subgraph GC["GC流程"]
        G1[GC扫描Dropped] --> G2[删除Binlog]
        G2 --> G3[删除Meta]
    end
    
    W6 --> F1
    F6 --> C1
    C6 --> G1
    
    style Write fill:#e3f2fd
    style Flush fill:#fff3e0
    style Compact fill:#f3e5f5
    style GC fill:#e8f5e9

stateDiagram-v2
    [*] --> Growing: AllocSegment
    Growing --> Sealed: SealSegment(满或超时)
    Sealed --> Flushing: DataNode开始Flush
    Flushing --> Flushed: Flush完成
    Flushed --> Indexed: 索引构建完成
    Indexed --> Dropped: DropSegment
    Dropped --> [*]: GC清理
    
    Growing --> Dropped: Collection被删除
    Sealed --> Dropped: Collection被删除
    
    note right of Growing
        状态：内存中
        操作：接受新数据写入
        大小：动态增长
    end note
    
    note right of Flushed
        状态：已持久化
        操作：只读
        位置：Object Storage
    end note

// 分配策略：优先复用未满Segment，不存在则创建新Segment
func (m *SegmentManager) AllocSegment(collectionID, partitionID int64, channelName string, count int64) ([]*Allocation, error) {
    // 1. 查找未满的Growing Segment
    segment := m.meta.GetGrowingSegment(collectionID, partitionID, channelName)
    
    // 2. 检查容量是否足够
    if segment != nil && segment.AvailableSize >= count {
        allocation := &Allocation{
            SegmentID: segment.ID,
            NumOfRows: count,
            ExpireTime: time.Now().Add(SegmentExpireDuration),
        }
        return []*Allocation{allocation}, nil
    }
    
    // 3. 创建新Segment
    newSegmentID, err := m.allocator.AllocID()
    if err != nil {
        return nil, err
    }
    
    segment = &SegmentInfo{
        SegmentID:    newSegmentID,
        CollectionID: collectionID,
        PartitionID:  partitionID,
        State:        commonpb.SegmentState_Growing,
        MaxSize:      SegmentMaxSize,  // 默认512MB
        Channel:      channelName,
    }
    
    m.meta.AddSegment(segment)
    
    return []*Allocation{{SegmentID: newSegmentID, NumOfRows: count}}, nil
}

flowchart LR
    C1[Collection1<br/>4 Channels] --> VC1[vchan_0]
    C1 --> VC2[vchan_1]
    C1 --> VC3[vchan_2]
    C1 --> VC4[vchan_3]
    
    VC1 -.->|Watch| DN1[DataNode1]
    VC2 -.->|Watch| DN2[DataNode2]
    VC3 -.->|Watch| DN1
    VC4 -.->|Watch| DN3[DataNode3]
    
    DN1 -->|Flush| S1[Segment1-3]
    DN2 -->|Flush| S2[Segment4-5]
    DN3 -->|Flush| S3[Segment6-7]
    
    style C1 fill:#e1f5ff
    style DN1 fill:#fff4e6
    style DN2 fill:#fff4e6
    style DN3 fill:#fff4e6

type meta struct {
    sync.RWMutex
    
    // Segment元数据：segmentID -> SegmentInfo
    segments map[UniqueID]*SegmentInfo
    
    // Collection -> Segments映射
    collections map[UniqueID][]UniqueID
    
    // Channel -> Segments映射
    channelSegments map[string][]UniqueID
    
    // 元数据持久化层
    catalog metastore.DataCoordCatalog
    
    // Collection信息缓存
    collectionInfos map[UniqueID]*collectionInfo
}

flowchart TD
    Start[扫描Segment] --> CheckDelete{Delete比例>20%}
    CheckDelete -->|是| MixCompaction[MixCompaction<br/>合并Delete Delta]
    CheckDelete -->|否| CheckSize{小文件数量>10}
    
    CheckSize -->|是| MergeCompaction[MergeCompaction<br/>合并小Segment]
    CheckSize -->|否| CheckCluster{是否启用Clustering}
    
    CheckCluster -->|是| ClusteringCompaction[ClusteringCompaction<br/>按标量字段聚类]
    CheckCluster -->|否| End[不触发]
    
    MixCompaction --> Schedule[提交到CompactionInspector]
    MergeCompaction --> Schedule
    ClusteringCompaction --> Schedule
    
    Schedule --> End
    
    style MixCompaction fill:#ffe6e6
    style MergeCompaction fill:#e6f3ff
    style ClusteringCompaction fill:#e6ffe6

// 计算Compaction收益分数
func (t *compactionTrigger) calculateScore(segments []*SegmentInfo) float64 {
    var totalSize int64
    var totalDeleted int64
    var fragmentScore float64
    
    for _, seg := range segments {
        totalSize += seg.Size
        totalDeleted += seg.NumOfRows * seg.DeleteRatio
        
        // 碎片化分数：小文件越多，分数越高
        if seg.Size < SmallSegmentThreshold {
            fragmentScore += 1.0
        }
    }
    
    deleteRatio := float64(totalDeleted) / float64(totalSize)
    
    // 综合分数 = 删除比例 * 0.7 + 碎片化分数 * 0.3
    return deleteRatio*0.7 + (fragmentScore/float64(len(segments)))*0.3
}

sequenceDiagram
    autonumber
    participant GC as GarbageCollector
    participant Meta as Meta
    participant OS as Object Storage
    
    loop 每30分钟
        GC->>Meta: 扫描Dropped Segment
        Meta-->>GC: [seg1, seg2, ...]
        
        loop 每个Segment
            GC->>Meta: 获取Segment的Binlog路径
            Meta-->>GC: [binlog1, binlog2, ...]
            
            GC->>OS: 检查文件是否存在
            OS-->>GC: 存在
            
            GC->>OS: 删除Binlog文件
            OS-->>GC: 删除成功
            
            GC->>Meta: 移除Segment元数据
            Meta-->>GC: 成功
        end
    end

sequenceDiagram
    autonumber
    participant P as Proxy
    participant DC as DataCoord::Server
    participant H as Handler
    participant SM as SegmentManager
    participant Meta as Meta
    participant Alloc as Allocator
    participant ETCD as etcd/TiKV
    
    P->>+DC: AssignSegmentID(requests)
    Note over DC: 检查健康状态
    
    loop 处理每个请求
        DC->>+H: GetCollection(collectionID)
        H->>RC: DescribeCollection(collectionID)
        RC-->>H: CollectionInfo(schema, properties)
        H-->>-DC: CollectionInfo
        
        DC->>+SM: AllocSegment(collID, partID, channel, count, version)
        
        Note over SM: 加Channel锁<br/>确保串行分配
        
        SM->>+Meta: GetGrowingSegments(channel)
        Note over Meta: 从内存缓存查询<br/>channel -> segments映射
        Meta-->>-SM: [segment1, segment2, ...]
        
        alt 存在可用Segment
            Note over SM: 按可用空间排序<br/>优先使用最满的Segment
            SM->>SM: 检查容量是否充足
            alt 容量足够
                SM->>Meta: 更新分配记录
                SM-->>DC: Allocation(existingSegID, count)
            else 容量不足需要新Segment
                SM->>+Alloc: AllocID()
                Alloc->>ETCD: IncrementBy(1)
                ETCD-->>Alloc: new_id
                Alloc-->>-SM: new_segment_id
                
                SM->>SM: 构建SegmentInfo
                Note over SM: state=Growing<br/>maxRows=1000000<br/>maxSize=512MB
                
                SM->>+Meta: AddSegment(newSegment)
                Meta->>Meta: 更新内存索引
                Note over Meta: 更新 segments map<br/>更新 channel2Growing map<br/>更新 collection2Segments map
                
                Meta->>+ETCD: SaveSegment(segmentInfo)
                Note over ETCD: key: segment/{segID}<br/>事务写入
                ETCD-->>-Meta: OK
                Meta-->>-SM: Success
                
                SM-->>DC: Allocation(newSegID, count)
            end
        else 不存在Growing Segment
            Note over SM: 同上创建新Segment流程
        end
        
        SM-->>-DC: [Allocation...]
        
        DC->>DC: 构建SegmentIDAssignment
        Note over DC: segmentID, count<br/>expireTime, collectionID
    end
    
    DC-->>-P: AssignSegmentIDResponse
    Note over P: 使用分配的SegmentID<br/>写入数据到DataNode

// internal/datacoord/services.go:307
func (s *Server) AssignSegmentID(ctx context.Context, req *datapb.AssignSegmentIDRequest) (*datapb.AssignSegmentIDResponse, error) {
    // 1. 健康检查
    if err := merr.CheckHealthy(s.GetStateCode()); err != nil {
        return &datapb.AssignSegmentIDResponse{
            Status: merr.Status(err),
        }, nil
    }
    
    // 2. 处理每个分配请求
    assigns := make([]*datapb.SegmentIDAssignment, 0, len(req.SegmentIDRequests))
    for _, r := range req.SegmentIDRequests {
        // 3. 获取Collection信息（从RootCoord）
        _, err := s.handler.GetCollection(ctx, r.GetCollectionID())
        
        // 4. 调用SegmentManager执行分配
        segmentAllocations, err := s.segmentManager.AllocSegment(ctx,
            r.CollectionID, r.PartitionID, r.ChannelName, 
            int64(r.Count), r.GetStorageVersion())
        
        // 5. 构造返回结果
        for _, allocation := range segmentAllocations {
            assigns = append(assigns, &datapb.SegmentIDAssignment{
                SegID:        allocation.SegmentID,
                ChannelName:  r.ChannelName,
                Count:        uint32(allocation.NumOfRows),
                CollectionID: r.CollectionID,
                PartitionID:  r.PartitionID,
                ExpireTime:   allocation.ExpireTime,
            })
        }
    }
    
    return &datapb.AssignSegmentIDResponse{
        Status:           merr.Success(),
        SegIDAssignments: assigns,
    }, nil
}

// internal/datacoord/segment_manager.go:240
func (sm *SegmentManager) AllocSegment(ctx context.Context, collectionID, partitionID UniqueID, 
    channelName string, requestRows int64, storageVersion int64) ([]*Allocation, error) {
    
    // 1. 加锁：确保同一个Channel的Segment分配串行化
    sm.channelLock.Lock(channelName)
    defer sm.channelLock.Unlock(channelName)
    
    // 2. 获取分配策略
    allocations, err := sm.allocPolicy.apply(collectionID, partitionID, 
        channelName, requestRows)
    
    // 3. 根据策略执行分配
    for _, alloc := range allocations {
        if alloc.SegmentID == 0 {
            // 需要创建新Segment
            segmentID, err := sm.allocator.allocID(ctx)
            
            segment := &SegmentInfo{
                SegmentID:     segmentID,
                CollectionID:  collectionID,
                PartitionID:   partitionID,
                InsertChannel: channelName,
                State:         commonpb.SegmentState_Growing,
                MaxRowNum:     Params.SegmentMaxSize,
                NumOfRows:     0,
                StorageVersion: storageVersion,
            }
            
            // 持久化到Meta
            err = sm.meta.AddSegment(ctx, segment)
            
            alloc.SegmentID = segmentID
        }
    }
    
    return allocations, nil
}

// internal/datacoord/meta.go:450
func (m *meta) AddSegment(ctx context.Context, segment *SegmentInfo) error {
    m.Lock()
    defer m.Unlock()
    
    // 1. 更新内存索引
    m.segments.Insert(segment.GetID(), segment)
    
    // 2. 更新Channel -> Segments映射
    if segment.State == commonpb.SegmentState_Growing {
        growingSet := m.channel2Growing.GetOrInsert(segment.InsertChannel, 
            typeutil.NewUniqueSet())
        growingSet.Insert(segment.GetID())
    }
    
    // 3. 更新Collection -> Segments映射
    collSegments := m.collections.GetOrInsert(segment.CollectionID, []int64{})
    m.collections.Insert(segment.CollectionID, append(collSegments, segment.GetID()))
    
    // 4. 持久化到etcd
    if err := m.catalog.AddSegment(ctx, segment.SegmentInfo); err != nil {
        // 回滚内存操作
        m.segments.Remove(segment.GetID())
        return err
    }
    
    return nil
}

// 使用Channel级别的锁，避免全局锁竞争
type SegmentManager struct {
    channelLock *lock.KeyLock[string]  // channel -> lock
    // ...
}

// 同一Channel的分配请求串行化
sm.channelLock.Lock(channelName)
defer sm.channelLock.Unlock(channelName)

// 优先复用未满的Growing Segment
func (sm *SegmentManager) selectOrCreateSegment(channel string, count int64) (*SegmentInfo, error) {
    segments := sm.meta.GetGrowingSegments(channel)
    
    // 按可用空间排序，优先选择最满的Segment（减少碎片）
    sort.Slice(segments, func(i, j int) bool {
        return segments[i].AvailableSize() < segments[j].AvailableSize()
    })
    
    for _, seg := range segments {
        if seg.AvailableSize() >= count {
            return seg, nil
        }
    }
    
    // 无可用Segment，创建新的
    return sm.createSegment(channel, count)
}

stateDiagram-v2
    [*] --> 内存更新: AddSegment
    内存更新 --> etcd持久化: Save
    etcd持久化 --> 成功: OK
    etcd持久化 --> 回滚内存: Error
    回滚内存 --> [*]: 失败
    成功 --> [*]: 完成
    
    note right of 内存更新
        • 更新 segments map
        • 更新 channel2Growing
        • 更新 collection2Segments
    end note
    
    note right of etcd持久化
        事务写入
        key: segment/{segID}
    end note

flowchart TD
    Start[AssignSegmentID] --> Check{健康检查}
    Check -->|失败| Err1[返回Unavailable]
    Check -->|成功| GetColl{获取Collection}
    
    GetColl -->|失败| Err2[返回CollectionNotFound]
    GetColl -->|成功| AllocSeg{分配Segment}
    
    AllocSeg -->|成功| Return[返回Assignment]
    AllocSeg -->|ID分配失败| Err3[返回OutOfID]
    AllocSeg -->|Meta持久化失败| Err4[返回InternalError]
    
    Err1 --> End
    Err2 --> End
    Err3 --> End
    Err4 --> End
    Return --> End[结束]
    
    style Err1 fill:#ffcdd2
    style Err2 fill:#ffcdd2
    style Err3 fill:#ffcdd2
    style Err4 fill:#ffcdd2

sequenceDiagram
    autonumber
    participant User as User/Client
    participant P as Proxy
    participant DC as DataCoord::Server
    participant SM as SegmentManager
    participant Meta as Meta
    participant CM as ChannelManager
    participant Cluster as Cluster
    participant DN as DataNode
    participant S3 as Object Storage
    participant QC as QueryCoord
    
    User->>+P: Flush(collectionID)
    P->>+DC: Flush(collectionID, segmentIDs)
    Note over DC: 手动触发Flush
    
    DC->>+DC: AllocTimestamp()
    DC-->>-DC: flushTs
    
    DC->>+DC: flushCollection(collID, flushTs, segIDs)
    
    DC->>+Meta: GetCollection(collID)
    Meta-->>-DC: CollectionInfo(channels)
    
    loop 每个VChannel
        DC->>+Meta: GetChannelCheckpoint(channel)
        Meta-->>-DC: checkpoint
        Note over DC: 记录当前Channel位置<br/>确保checkpoint早于segment endTs
        
        alt segIDs为空（Flush全部）
            DC->>+SM: SealAllSegments(channel, [])
            SM->>+Meta: GetSegmentsByChannel(channel)
            Meta-->>-SM: [growing_segments]
            
            loop 每个Growing Segment
                SM->>SM: 检查是否满足Seal条件
                Note over SM: size>threshold OR<br/>time>timeout
                
                alt 满足条件
                    SM->>+Meta: UpdateSegment(segID, state=Sealed)
                    Meta->>Meta: 更新内存状态
                    Meta->>ETCD: SaveSegment
                    ETCD-->>Meta: OK
                    Meta-->>-SM: Success
                end
            end
            SM-->>-DC: [sealed_segment_ids]
        else 指定segmentIDs
            DC->>+SM: SealAllSegments(channel, segIDs)
            Note over SM: 只Seal指定的Segments
            SM-->>-DC: [sealed_segment_ids]
        end
        
        DC->>+CM: GetNodeChannelsByCollectionID(collID)
        CM-->>-DC: nodeChannels{nodeID -> channels}
        
        loop 每个DataNode
            DC->>+Cluster: FlushChannels(nodeID, flushTs, channels)
            Cluster->>+DN: FlushChannels RPC
            
            Note over DN: 异步执行Flush<br/>写入Binlog
            
            DN->>DN: 遍历所有Sealed Segment
            
            loop 每个Segment
                DN->>DN: 序列化Insert数据
                DN->>DN: 序列化Delete数据
                DN->>DN: 生成Stats Log
                
                DN->>+S3: PutObject(insert_log)
                S3-->>-DN: binlog_path
                
                DN->>+S3: PutObject(delete_log)
                S3-->>-DN: deltalog_path
                
                DN->>+S3: PutObject(stats_log)
                S3-->>-DN: statslog_path
            end
            
            DN->>+DC: SegmentFlushCompleted(segmentID, binlogs)
            DC->>+Meta: UpdateSegmentBinlogs(segID, binlogs)
            Meta->>Meta: 更新Binlog路径
            Meta->>+ETCD: SaveSegment(binlogs)
            ETCD-->>-Meta: OK
            Meta-->>-DC: Success
            
            DC->>+Meta: UpdateSegmentState(segID, Flushed)
            Meta-->>-DC: Success
            DC-->>-DN: ACK
            
            DN-->>-Cluster: FlushCompleted
            Cluster-->>-DC: Success
        end
        
        DC->>+QC: SegmentFlushCompleted(segments)
        Note over QC: 触发Handoff<br/>加载Segment到QueryNode
        QC-->>-DC: ACK
    end
    
    DC->>DC: 收集Flush结果
    DC-->>-DC: FlushResult
    DC-->>-P: FlushResponse(segmentIDs, flushTs)
    P-->>-User: Success

// internal/datacoord/services.go:77
func (s *Server) Flush(ctx context.Context, req *datapb.FlushRequest) (*datapb.FlushResponse, error) {
    // 1. 健康检查
    if err := merr.CheckHealthy(s.GetStateCode()); err != nil {
        return &datapb.FlushResponse{Status: merr.Status(err)}, nil
    }
    
    // 2. 分配Flush时间戳（全局顺序保证）
    ts, err := s.allocator.AllocTimestamp(ctx)
    if err != nil {
        return nil, err
    }
    
    // 3. 执行Collection级Flush
    flushResult, err := s.flushCollection(ctx, req.GetCollectionID(), ts, req.GetSegmentIDs())
    if err != nil {
        return &datapb.FlushResponse{Status: merr.Status(err)}, nil
    }
    
    return &datapb.FlushResponse{
        Status:          merr.Success(),
        SegmentIDs:      flushResult.GetSegmentIDs(),      // Sealed的Segment
        FlushSegmentIDs: flushResult.GetFlushSegmentIDs(), // 已Flushed的Segment
        TimeOfSeal:      flushResult.GetTimeOfSeal(),
        FlushTs:         flushResult.GetFlushTs(),
        ChannelCps:      flushResult.GetChannelCps(),      // Channel Checkpoint
    }, nil
}

// internal/datacoord/services.go:116
func (s *Server) flushCollection(ctx context.Context, collectionID UniqueID, 
    flushTs uint64, toFlushSegments []UniqueID) (*datapb.FlushResult, error) {
    
    // 1. 获取Collection信息
    coll, err := s.handler.GetCollection(ctx, collectionID)
    if err != nil {
        return nil, err
    }
    
    // 2. 获取所有Channel的Checkpoint（必须在Seal之前）
    //    确保checkpoint时间早于segment的endTs
    channelCPs := make(map[string]*msgpb.MsgPosition, 0)
    for _, vchannel := range coll.VChannelNames {
        cp := s.meta.GetChannelCheckpoint(vchannel)
        channelCPs[vchannel] = cp
    }
    
    // 3. Seal所有需要Flush的Segment
    sealedSegmentsIDDict := make(map[UniqueID]bool)
    for _, channel := range coll.VChannelNames {
        sealedSegmentIDs, err := s.segmentManager.SealAllSegments(ctx, channel, toFlushSegments)
        if err != nil {
            return nil, errors.Wrapf(err, "failed to flush collection %d", collectionID)
        }
        for _, sealedSegmentID := range sealedSegmentIDs {
            sealedSegmentsIDDict[sealedSegmentID] = true
        }
    }
    
    // 4. 获取所有已Flushed的Segment（用于返回给用户）
    segments := s.meta.GetSegmentsOfCollection(ctx, collectionID)
    flushSegmentIDs := make([]UniqueID, 0, len(segments))
    for _, segment := range segments {
        if segment != nil && 
           isFlushState(segment.GetState()) && 
           segment.GetLevel() != datapb.SegmentLevel_L0 {
            flushSegmentIDs = append(flushSegmentIDs, segment.GetID())
        }
    }
    
    // 5. 通知DataNode执行Flush
    nodeChannels := s.channelManager.GetNodeChannelsByCollectionID(collectionID)
    for nodeID, channelNames := range nodeChannels {
        err = s.cluster.FlushChannels(ctx, nodeID, flushTs, channelNames)
        if err != nil {
            return nil, err
        }
    }
    
    return &datapb.FlushResult{
        CollectionID:    collectionID,
        SegmentIDs:      lo.Keys(sealedSegmentsIDDict),
        FlushSegmentIDs: flushSegmentIDs,
        FlushTs:         flushTs,
        ChannelCps:      channelCPs,
    }, nil
}

// internal/datacoord/segment_manager.go:450
func (sm *SegmentManager) SealAllSegments(ctx context.Context, channel string, 
    segIDs []UniqueID) ([]UniqueID, error) {
    
    sm.channelLock.Lock(channel)
    defer sm.channelLock.Unlock(channel)
    
    // 1. 获取Channel的所有Growing Segment
    segments := sm.meta.GetGrowingSegments(channel)
    
    // 2. 过滤需要Seal的Segment
    toSeal := []UniqueID{}
    if len(segIDs) > 0 {
        // 指定Segment
        segIDSet := typeutil.NewUniqueSet(segIDs...)
        for _, seg := range segments {
            if segIDSet.Contain(seg.GetID()) {
                toSeal = append(toSeal, seg.GetID())
            }
        }
    } else {
        // 所有Growing Segment
        for _, seg := range segments {
            toSeal = append(toSeal, seg.GetID())
        }
    }
    
    // 3. 执行Seal操作
    sealed := []UniqueID{}
    for _, segID := range toSeal {
        segment := sm.meta.GetSegment(ctx, segID)
        if segment == nil || segment.GetState() != commonpb.SegmentState_Growing {
            continue
        }
        
        // 更新状态为Sealed
        segment.State = commonpb.SegmentState_Sealed
        if err := sm.meta.UpdateSegment(ctx, segment); err != nil {
            return nil, err
        }
        
        sealed = append(sealed, segID)
        
        // 从Growing索引移除
        sm.channel2Growing.GetOrInsert(channel, typeutil.NewUniqueSet()).Remove(segID)
        // 加入Sealed索引
        sm.channel2Sealed.GetOrInsert(channel, typeutil.NewUniqueSet()).Insert(segID)
    }
    
    return sealed, nil
}

// internal/datacoord/services.go:1900
func (s *Server) SegmentFlushCompleted(ctx context.Context, 
    req *datapb.SegmentFlushCompletedRequest) (*commonpb.Status, error) {
    
    segmentID := req.GetSegment().GetID()
    
    // 1. 更新Segment的Binlog路径
    segment := s.meta.GetSegment(ctx, segmentID)
    if segment == nil {
        return merr.Status(merr.WrapErrSegmentNotFound(segmentID)), nil
    }
    
    // 2. 更新Binlog、Deltalog、Statslog
    segment.Binlogs = req.GetSegment().GetBinlogs()
    segment.Deltalogs = req.GetSegment().GetDeltalogs()
    segment.Statslogs = req.GetSegment().GetStatslogs()
    segment.NumOfRows = req.GetSegment().GetNumOfRows()
    
    // 3. 更新状态为Flushed
    segment.State = commonpb.SegmentState_Flushed
    segment.DmlPosition = req.GetSegment().GetDmlPosition()
    
    // 4. 持久化
    if err := s.meta.UpdateSegment(ctx, segment); err != nil {
        return merr.Status(err), nil
    }
    
    // 5. 通知QueryCoord Handoff
    s.notifyQueryCoordSegmentFlush(ctx, segment)
    
    return merr.Success(), nil
}

# 方式1：Flush指定Collection
client.flush(collection_name="my_collection")

# 方式2：Flush指定Segment
client.flush(collection_name="my_collection", segment_ids=[seg1, seg2])

# 方式3：FlushAll（所有Collection）
# 管理员操作

stateDiagram-v2
    [*] --> Growing: 创建Segment
    Growing --> Sealed: Flush触发
    Sealed --> Flushing: DataNode开始Flush
    Flushing --> Flushed: Binlog写入完成
    Flushed --> Indexed: 索引构建完成
    Indexed --> [*]: 可被查询
    
    Growing --> Dropped: Collection删除
    Sealed --> Dropped: Collection删除
    Flushing --> Dropped: Collection删除
    
    note right of Growing
        状态：内存中
        特征：
        • 接受新数据写入
        • 大小动态增长
        • NumOfRows递增
    end note
    
    note right of Sealed
        状态：已封闭
        特征：
        • 不再接受新数据
        • 等待DataNode Flush
        • DmlPosition已确定
    end note
    
    note right of Flushed
        状态：已持久化
        特征：
        • Binlog已写入S3
        • 只读
        • 可被QueryNode加载
    end note

flowchart TD
    Start[收到FlushChannels命令] --> Loop{遍历Sealed Segment}
    
    Loop -->|有Segment| GetData[从Insert Buffer获取数据]
    GetData --> Serialize[序列化数据]
    
    Serialize --> WriteInsert[写Insert Log]
    WriteInsert --> WriteDelete[写Delete Log]
    WriteDelete --> WriteStats[写Stats Log]
    
    WriteStats --> CalcSize[计算Segment大小]
    CalcSize --> Report[上报FlushCompleted]
    
    Report --> Loop
    Loop -->|无Segment| End[完成]
    
    style WriteInsert fill:#e3f2fd
    style WriteDelete fill:#fff3e0
    style WriteStats fill:#f3e5f5

├── Insert Log (列式存储)
│   ├── field_0/  (FieldID=0, 主键)
│   │   └── 435978159261483010
│   ├── field_1/  (FieldID=1, 向量字段)
│   │   └── 435978159261483011
│   └── field_2/  (FieldID=2, 标量字段)
│       └── 435978159261483012
│
├── Delete Log (删除记录)
│   └── 435978159261483020
│
└── Stats Log (统计信息)
    ├── 435978159261483030 (min/max/count)
    └── 435978159261483031

flowchart TD
    Start[Flush请求] --> Check1{Collection存在?}
    Check1 -->|否| Err1[返回CollectionNotFound]
    Check1 -->|是| Check2{Segment有效?}
    
    Check2 -->|否| Err2[返回SegmentNotFound]
    Check2 -->|是| Seal{Seal Segment}
    
    Seal -->|成功| Notify{通知DataNode}
    Seal -->|失败| Err3[返回Meta更新失败]
    
    Notify -->|成功| Wait[等待DataNode完成]
    Notify -->|超时| Err4[返回Timeout<br/>Segment仍会后台Flush]
    
    Wait -->|完成| Update{更新Meta}
    Update -->|成功| Success[返回Success]
    Update -->|失败| Err5[返回Meta更新失败<br/>Binlog已写入<br/>后台重试]
    
    style Err1 fill:#ffcdd2
    style Err2 fill:#ffcdd2
    style Err3 fill:#ffcdd2
    style Err4 fill:#fff9c4
    style Err5 fill:#fff9c4
    style Success fill:#c8e6c9

sequenceDiagram
    autonumber
    participant CTM as CompactionTriggerManager
    participant Policy as CompactionPolicy
    participant CI as CompactionInspector
    participant Meta as Meta
    participant Cluster as Cluster
    participant DN as DataNode
    participant S3 as Object Storage
    participant QC as QueryCoord
    
    Note over CTM: 定时触发（每10分钟）
    
    CTM->>+CTM: loop() - 后台任务
    
    CTM->>+Policy: Trigger(ctx)
    Note over Policy: 扫描Collection<br/>评估Compaction收益
    
    Policy->>+Meta: GetSegmentsOfCollection(collID)
    Meta-->>-Policy: [segments]
    
    Policy->>Policy: 按类型生成CompactionView
    Note over Policy: • L0 Compaction<br/>• Mix Compaction<br/>• Clustering
    
    Policy->>Policy: 计算收益分数
    Note over Policy: DeleteRatio * 0.4<br/>+ FragmentScore * 0.3<br/>+ AgeScore * 0.2
    
    alt 分数 > 阈值
        Policy-->>-CTM: [CompactionViews]
        
        loop 每个CompactionView
            CTM->>CTM: triggerViewForCompaction()
            
            CTM->>+CI: SubmitTask(CompactionTask)
            Note over CI: 加入任务队列
            
            CI->>+Meta: GetSegmentDetails(segmentIDs)
            Meta-->>-CI: [SegmentInfo with binlogs]
            
            CI->>CI: 构建CompactionPlan
            Note over CI: 选择DataNode<br/>分配资源
            
            CI->>+Cluster: Compaction(nodeID, plan)
            Cluster->>+DN: Compaction RPC
            
            Note over DN: 异步执行Compaction
            
            DN->>DN: 创建CompactionTask
            
            loop 每个输入Segment
                DN->>+S3: GetObject(insert_binlog)
                S3-->>-DN: binlog_data
                
                DN->>+S3: GetObject(delta_binlog)
                S3-->>-DN: delta_data
                
                DN->>DN: 反序列化数据
                DN->>DN: 应用Delete操作
            end
            
            DN->>DN: 合并数据
            Note over DN: • 去重<br/>• 删除已标记删除的行<br/>• 按PK排序
            
            alt Mix Compaction
                Note over DN: N个Segment -> 1个大Segment
                DN->>DN: 合并所有数据到一个Segment
            else L0 Compaction
                Note over DN: L0 Delta -> L1 Segment
                DN->>DN: 将L0删除应用到L1
            else Clustering Compaction
                Note over DN: 按标量字段聚类
                DN->>DN: 按ClusteringKey重新分组
            end
            
            DN->>DN: 序列化新Segment
            
            DN->>+S3: PutObject(new_insert_binlog)
            S3-->>-DN: new_binlog_path
            
            DN->>+S3: PutObject(new_stats_binlog)
            S3-->>-DN: new_statslog_path
            
            DN->>+DC: CompactionCompleted(result)
            DC->>+CI: HandleCompactionResult(result)
            
            CI->>+Meta: AddSegment(newSegment)
            Meta->>Meta: 更新内存索引
            Meta->>+ETCD: SaveSegment
            ETCD-->>-Meta: OK
            Meta-->>-CI: Success
            
            loop 每个旧Segment
                CI->>+Meta: DropSegment(oldSegID)
                Meta->>Meta: 标记state=Dropped
                Meta->>+ETCD: SaveSegment(state=Dropped)
                ETCD-->>-Meta: OK
                Meta-->>-CI: Success
            end
            
            CI->>+QC: NotifySegmentChange(newSegment, oldSegments)
            Note over QC: 触发Segment替换<br/>先加载新Segment<br/>再释放旧Segment
            QC-->>-CI: ACK
            
            CI-->>-DC: Success
            DC-->>-DN: ACK
            DN-->>-Cluster: CompactionCompleted
            Cluster-->>-CI: Success
            
            CI->>CI: 更新任务状态
            Note over CI: state=Completed
            CI-->>-CTM: TaskCompleted
        end
    else 分数 <= 阈值
        Policy-->>CTM: nil (不触发)
    end
    
    CTM-->>-CTM: 继续下一轮循环

flowchart LR
    subgraph Input["输入"]
        L0[L0 Segment<br/>━━━━━━<br/>Delete Records<br/>PK: 1,3,5,7]
        L1[L1 Segment<br/>━━━━━━<br/>Data Records<br/>PK: 1,2,3,4,5,6,7,8]
    end
    
    subgraph Process["处理"]
        Apply[应用Delete操作<br/>━━━━━━<br/>移除 PK=1,3,5,7]
    end
    
    subgraph Output["输出"]
        NewL1[New L1 Segment<br/>━━━━━━<br/>PK: 2,4,6,8]
        DropL0[Drop L0<br/>标记删除]
        DropOldL1[Drop Old L1<br/>标记删除]
    end
    
    L0 --> Apply
    L1 --> Apply
    Apply --> NewL1
    Apply --> DropL0
    Apply --> DropOldL1
    
    style L0 fill:#ffebee
    style L1 fill:#e3f2fd
    style NewL1 fill:#c8e6c9

flowchart LR
    subgraph Input["输入"]
        S1[Segment1<br/>64MB]
        S2[Segment2<br/>128MB]
        S3[Segment3<br/>96MB]
        S4[Segment4<br/>80MB]
    end
    
    subgraph Process["处理"]
        Merge[合并数据<br/>━━━━━━<br/>• 去重<br/>• 应用Delete<br/>• 按PK排序]
    end
    
    subgraph Output["输出"]
        NewS[New Segment<br/>━━━━━━<br/>368MB<br/>已优化]
    end
    
    S1 --> Merge
    S2 --> Merge
    S3 --> Merge
    S4 --> Merge
    Merge --> NewS
    
    style Input fill:#fff3e0
    style NewS fill:#c8e6c9

flowchart TB
    subgraph Input["输入数据"]
        Data[原始Segment<br/>━━━━━━<br/>随机分布<br/>ClusteringKey无序]
    end
    
    subgraph Process["处理"]
        Analyze[分析ClusteringKey<br/>━━━━━━<br/>计算分区边界]
        Partition[按Key分区<br/>━━━━━━<br/>Range Partitioning]
    end
    
    subgraph Output["输出"]
        C1[Cluster1<br/>key: 0-100]
        C2[Cluster2<br/>key: 100-200]
        C3[Cluster3<br/>key: 200-300]
    end
    
    Data --> Analyze
    Analyze --> Partition
    Partition --> C1
    Partition --> C2
    Partition --> C3
    
    style C1 fill:#e8f5e9
    style C2 fill:#e8f5e9
    style C3 fill:#e8f5e9

// internal/datacoord/compaction_trigger_v2.go:212
func (m *CompactionTriggerManager) loop(ctx context.Context) {
    // 定时器
    l0Ticker := time.NewTicker(Params.DataCoordCfg.L0CompactionTriggerInterval.GetAsDuration(time.Second))
    clusteringTicker := time.NewTicker(Params.DataCoordCfg.ClusteringCompactionTriggerInterval.GetAsDuration(time.Second))
    singleTicker := time.NewTicker(Params.DataCoordCfg.MixCompactionTriggerInterval.GetAsDuration(time.Second))
    
    for {
        select {
        case <-ctx.Done():
            return
            
        case <-l0Ticker.C:
            if !m.l0Policy.Enable() {
                continue
            }
            if m.inspector.isFull() {
                log.RatedInfo(10, "Skip trigger l0 compaction since inspector is full")
                continue
            }
            
            // 触发L0 Compaction
            events, err := m.l0Policy.Trigger(ctx)
            if err != nil {
                log.Warn("Fail to trigger L0 policy", zap.Error(err))
                continue
            }
            
            // 提交任务
            for triggerType, views := range events {
                m.notify(ctx, triggerType, views)
            }
            
        case <-clusteringTicker.C:
            // 触发Clustering Compaction
            events, err := m.clusteringPolicy.Trigger(ctx)
            for triggerType, views := range events {
                m.notify(ctx, triggerType, views)
            }
            
        case <-singleTicker.C:
            // 触发Mix Compaction
            events, err := m.singlePolicy.Trigger(ctx)
            for triggerType, views := range events {
                m.notify(ctx, triggerType, views)
            }
        }
    }
}

// internal/datacoord/compaction_policy_single.go:100
func (policy *singleCompactionPolicy) Trigger(ctx context.Context) (map[CompactionTriggerType][]CompactionView, error) {
    collections := policy.meta.GetCollections()
    views := make([]CompactionView, 0)
    
    for _, collectionID := range collections {
        // 获取所有Segment
        segments := policy.meta.GetSegmentsOfCollection(ctx, collectionID)
        
        // 按Channel分组
        channelSegments := groupSegmentsByChannel(segments)
        
        for channel, segs := range channelSegments {
            // 计算Compaction收益
            score := policy.calculateScore(segs)
            
            if score > policy.threshold {
                // 生成CompactionView
                view := &compactionView{
                    collectionID: collectionID,
                    channel:      channel,
                    segments:     segs,
                    triggerID:    allocator.AllocID(),
                }
                views = append(views, view)
            }
        }
    }
    
    result := make(map[CompactionTriggerType][]CompactionView)
    if len(views) > 0 {
        result[TriggerTypeSingle] = views
    }
    return result, nil
}

// 计算Compaction收益分数
func (policy *singleCompactionPolicy) calculateScore(segments []*SegmentInfo) float64 {
    var totalSize, totalRows, deletedRows int64
    var smallFileCount int
    
    for _, seg := range segments {
        totalSize += seg.Size
        totalRows += seg.NumRows
        deletedRows += seg.DeletedRows
        
        if seg.Size < SmallSegmentThreshold {  // 256MB
            smallFileCount++
        }
    }
    
    deleteRatio := float64(deletedRows) / float64(totalRows)
    fragmentScore := float64(smallFileCount) / float64(len(segments))
    ageScore := float64(time.Since(segments[0].LastCompactionTime).Hours()) / (30 * 24)
    sizeScore := float64(totalSize) / (10 * 1024 * 1024 * 1024)  // 10GB
    
    // 加权计算
    return deleteRatio*0.4 + fragmentScore*0.3 + ageScore*0.2 + sizeScore*0.1
}

// internal/datacoord/compaction_inspector.go:200
func (ci *CompactionInspector) SubmitTask(task *CompactionTask) error {
    // 1. 检查任务队列是否满
    if ci.queue.IsFull() {
        return errors.New("compaction queue is full")
    }
    
    // 2. 选择DataNode
    nodeID, err := ci.selectDataNode(task)
    if err != nil {
        return err
    }
    
    // 3. 加入任务队列
    task.NodeID = nodeID
    task.State = datapb.CompactionTaskState_Pending
    ci.queue.Enqueue(task)
    
    // 4. 持久化任务状态
    ci.meta.SaveCompactionTask(task)
    
    // 5. 通知Scheduler调度
    ci.scheduler.Schedule(task)
    
    return nil
}

# Compaction任务数
milvus_datacoord_compaction_task_num:
  labels: [state, type]  # Pending/Running/Completed/Failed
  
# Compaction延迟
milvus_datacoord_compaction_latency:
  labels: [type]  # L0/Mix/Clustering
  unit: seconds
  percentiles: [p50, p90, p99]
  
# Compaction吞吐
milvus_datacoord_compaction_throughput:
  labels: [type]
  unit: bytes_per_second
  
# Compaction收益
milvus_datacoord_compaction_benefit:
  labels: [type]
  description: 节省的存储空间或查询加速比

sequenceDiagram
    autonumber
    participant RC as RootCoord
    participant DC as DataCoord::Server
    participant CM as ChannelManager
    participant Store as ChannelStore
    participant Policy as AssignPolicy
    participant Cluster as Cluster
    participant DN as DataNode
    participant ETCD as etcd
    
    RC->>+DC: WatchChannels(collID, channels, startPos)
    Note over RC: 创建Collection时触发
    
    DC->>DC: 健康检查
    
    loop 每个Channel
        DC->>DC: NewRWChannel(channelName, collID, startPos)
        Note over DC: 构建Channel对象<br/>包含schema、dbProperties
        
        DC->>+CM: Watch(ctx, channel)
        Note over CM: 加锁确保串行
        
        CM->>+Store: AddChannel(bufferID, channel)
        Note over Store: 先加入Buffer<br/>等待分配给DataNode
        
        Store->>+ETCD: SaveChannelWatchInfo
        Note over ETCD: key: channel/{channelName}<br/>state: ToWatch<br/>opID: xxx
        ETCD-->>-Store: OK
        Store-->>-CM: Success
        
        CM->>+Policy: apply(nodesChannels, bufferChannel)
        Note over Policy: 选择DataNode<br/>负载均衡算法
        
        Policy->>Policy: 计算每个Node负载
        Policy->>Policy: 选择负载最低的Node
        Policy-->>-CM: ChannelOpSet(nodeID, Watch, channel)
        
        CM->>+Store: UpdateChannelWatchInfo
        Note over Store: 更新state: ToWatch -> Watching
        Store->>+ETCD: SaveChannelWatchInfo(nodeID, state)
        ETCD-->>-Store: OK
        Store-->>-CM: Success
        
        CM->>+Cluster: NotifyChannelOperation(nodeID, watchInfo)
        
        Cluster->>+DN: WatchDmChannels RPC
        Note over DN: DataNode开始订阅消息
        
        DN->>DN: 创建FlowGraph
        DN->>DN: 订阅MQ Channel
        DN->>DN: Seek到startPosition
        
        DN-->>-Cluster: Success
        Cluster-->>-CM: Success
        
        CM->>+Store: UpdateChannelWatchInfo(state=Watched)
        Store->>+ETCD: SaveChannelWatchInfo
        ETCD-->>-Store: OK
        Store-->>-CM: Success
        
        CM-->>-DC: Success
        
        DC->>+ETCD: MarkChannelAdded(channelName)
        Note over ETCD: 标记Channel已添加<br/>防止重复创建
        ETCD-->>-DC: OK
        
        DC->>+DC: UpdateChannelCheckpoint(channel, startPos)
        Note over DC: 初始化Channel Checkpoint
        DC-->>-DC: Success
    end
    
    DC-->>-RC: WatchChannelsResponse(Success)