milvus Delete api写s3的流程

dataNode:
  segment:
    insertBufSize: 16777216 # Max buffer size to flush for a single segment.
    deleteBufBytes: 67108864 # Max buffer size to flush del for a single channel
    syncPeriod: 600 # The period to sync segments if buffer is not empty.

start()(internal\util\flowgraph\node.go)
  |--go nodeCtx.work()(同上)
    |--n.Operate(input)(同上)
      |--dn.bufferDeleteMsg()(internal\datanode\flow_graph_delete_node.go)
        |--dn.delBufferManager.StoreNewDeletes()(同上)

func (m *DeltaBufferManager) StoreNewDeletes(segID UniqueID, pks []primaryKey,
	tss []Timestamp, tr TimeRange, startPos, endPos *msgpb.MsgPosition,
) {
    // 获取delDataBuf
	buffer, loaded := m.Load(segID)
    // 如果不存在则新建
	if !loaded {
		buffer = newDelDataBuf(segID)
	}
    // 将pks存入buffer
	size := buffer.Buffer(pks, tss, tr, startPos, endPos)

	m.pushOrFixHeap(segID, buffer)
	m.updateMeta(segID, buffer)
	m.usedMemory.Add(size)

	metrics.DataNodeConsumeMsgRowsCount.WithLabelValues(
		fmt.Sprint(paramtable.GetNodeID()), metrics.DeleteLabel).Add(float64(len(pks)))
}

Start()(internal\util\flowgraph\node.go)
  |--go nodeCtx.work()(同上)
    |--n.Operate(input)(同上)
      |--dn.flushManager.flushDelData()(internal\datanode\flow_graph_delete_node.go)
        |--m.handleDeleteTask()(internal\datanode\flush_manager.go)
          |--m.getFlushQueue(segmentID).enqueueDelFlush()(同上)
            |--q.getFlushTaskRunner(pos).runFlushDel()(同上)
              |--runFlushDel()(internal\datanode\flush_task.go)
                |--task.flushDeleteData()(同上)
                  |--t.MultiWrite(ctx, t.data)(internal\datanode\flush_manager.go)

// notify flush manager del buffer data
func (m *rendezvousFlushManager) flushDelData(data *DelDataBuf, segmentID UniqueID,
	pos *msgpb.MsgPosition,
) error {
	// del signal with empty data
	if data == nil || data.delData == nil {
		m.handleDeleteTask(segmentID, &flushBufferDeleteTask{}, nil, pos)
		return nil
	}

	collID, partID, err := m.getCollectionAndPartitionID(segmentID)
	if err != nil {
		return err
	}
    // 编码解码器,提供序列化,反序列化功能
	delCodec := storage.NewDeleteCodec()
    // 序列化
	blob, err := delCodec.Serialize(collID, partID, segmentID, data.delData)
	if err != nil {
		return err
	}

	logID, err := m.AllocOne()
	if err != nil {
		log.Error("failed to alloc ID", zap.Error(err))
		return err
	}

	blobKey := metautil.JoinIDPath(collID, partID, segmentID, logID)
	blobPath := path.Join(m.ChunkManager.RootPath(), common.SegmentDeltaLogPath, blobKey)
    // 合成kvs
	kvs := map[string][]byte{blobPath: blob.Value[:]}
	data.LogSize = int64(len(blob.Value))
	data.LogPath = blobPath
	log.Info("delete blob path", zap.String("path", blobPath))
	m.handleDeleteTask(segmentID, &flushBufferDeleteTask{
		ChunkManager: m.ChunkManager,
		data:         kvs,
	}, data, pos)
	return nil
}

// Blob is a pack of key&value
type Blob struct {
	Key    string
	Value  []byte
	Size   int64
	RowNum int64
}

files/delta_log/447918772784340274/447918772784340275/447918772784340283/447918772783939606