milvus/internal/proto/data_coord.proto

syntax = "proto3";

package milvus.proto.data;

option go_package = "github.com/milvus-io/milvus/internal/proto/datapb";

import "common.proto";
import "internal.proto";
import "milvus.proto";
import "schema.proto";

// TODO: import google/protobuf/empty.proto
message Empty {}

enum SegmentType {
  New = 0;
  Normal = 1;
  Flushed = 2;
  Compacted = 3;
}

service DataCoord {
  rpc GetComponentStates(milvus.GetComponentStatesRequest) returns (milvus.ComponentStates) {}
  rpc GetTimeTickChannel(internal.GetTimeTickChannelRequest) returns(milvus.StringResponse) {}
  rpc GetStatisticsChannel(internal.GetStatisticsChannelRequest) returns(milvus.StringResponse){}

  rpc Flush(FlushRequest) returns (FlushResponse) {}

  rpc AssignSegmentID(AssignSegmentIDRequest) returns (AssignSegmentIDResponse) {}

  rpc GetSegmentInfo(GetSegmentInfoRequest) returns (GetSegmentInfoResponse) {}
  rpc GetSegmentStates(GetSegmentStatesRequest) returns (GetSegmentStatesResponse) {}
  rpc GetInsertBinlogPaths(GetInsertBinlogPathsRequest) returns (GetInsertBinlogPathsResponse) {}

  rpc GetCollectionStatistics(GetCollectionStatisticsRequest) returns (GetCollectionStatisticsResponse) {}
  rpc GetPartitionStatistics(GetPartitionStatisticsRequest) returns (GetPartitionStatisticsResponse) {}

  rpc GetSegmentInfoChannel(GetSegmentInfoChannelRequest) returns (milvus.StringResponse){}

  rpc SaveBinlogPaths(SaveBinlogPathsRequest) returns (common.Status){}
  rpc GetRecoveryInfo(GetRecoveryInfoRequest) returns (GetRecoveryInfoResponse){}
  rpc GetFlushedSegments(GetFlushedSegmentsRequest) returns(GetFlushedSegmentsResponse){}
  rpc GetSegmentsByStates(GetSegmentsByStatesRequest) returns(GetSegmentsByStatesResponse){}

  rpc ShowConfigurations(internal.ShowConfigurationsRequest) returns (internal.ShowConfigurationsResponse){}
  // https://wiki.lfaidata.foundation/display/MIL/MEP+8+--+Add+metrics+for+proxy
  rpc GetMetrics(milvus.GetMetricsRequest) returns (milvus.GetMetricsResponse) {}
  rpc ManualCompaction(milvus.ManualCompactionRequest) returns (milvus.ManualCompactionResponse) {}
  rpc GetCompactionState(milvus.GetCompactionStateRequest) returns (milvus.GetCompactionStateResponse) {}
  rpc GetCompactionStateWithPlans(milvus.GetCompactionPlansRequest) returns (milvus.GetCompactionPlansResponse) {}

  rpc WatchChannels(WatchChannelsRequest) returns (WatchChannelsResponse) {}
  rpc GetFlushState(milvus.GetFlushStateRequest) returns (milvus.GetFlushStateResponse) {}
  rpc DropVirtualChannel(DropVirtualChannelRequest) returns (DropVirtualChannelResponse) {}

  rpc SetSegmentState(SetSegmentStateRequest) returns (SetSegmentStateResponse) {}
  // https://wiki.lfaidata.foundation/display/MIL/MEP+24+--+Support+bulk+load
  rpc Import(ImportTaskRequest) returns (ImportTaskResponse) {}
  rpc UpdateSegmentStatistics(UpdateSegmentStatisticsRequest) returns (common.Status) {}
  rpc UpdateChannelCheckpoint(UpdateChannelCheckpointRequest) returns (common.Status) {}

  rpc AcquireSegmentLock(AcquireSegmentLockRequest) returns (common.Status) {}
  rpc ReleaseSegmentLock(ReleaseSegmentLockRequest) returns (common.Status) {}

  rpc SaveImportSegment(SaveImportSegmentRequest) returns(common.Status) {}
  rpc UnsetIsImportingState(UnsetIsImportingStateRequest) returns(common.Status) {}
  rpc MarkSegmentsDropped(MarkSegmentsDroppedRequest) returns(common.Status) {}

  rpc BroadcastAlteredCollection(AlterCollectionRequest) returns (common.Status) {}

  rpc CheckHealth(milvus.CheckHealthRequest) returns (milvus.CheckHealthResponse) {}
}

service DataNode {
  rpc GetComponentStates(milvus.GetComponentStatesRequest) returns (milvus.ComponentStates) {}
  rpc GetStatisticsChannel(internal.GetStatisticsChannelRequest) returns (milvus.StringResponse) {}

  rpc WatchDmChannels(WatchDmChannelsRequest) returns (common.Status) {}
  rpc FlushSegments(FlushSegmentsRequest) returns(common.Status) {}

  rpc ShowConfigurations(internal.ShowConfigurationsRequest) returns (internal.ShowConfigurationsResponse){}
  // https://wiki.lfaidata.foundation/display/MIL/MEP+8+--+Add+metrics+for+proxy
  rpc GetMetrics(milvus.GetMetricsRequest) returns (milvus.GetMetricsResponse) {}

  rpc Compaction(CompactionPlan) returns (common.Status) {}
  rpc GetCompactionState(CompactionStateRequest) returns (CompactionStateResponse) {}
  rpc SyncSegments(SyncSegmentsRequest) returns (common.Status) {}

  // https://wiki.lfaidata.foundation/display/MIL/MEP+24+--+Support+bulk+load
  rpc Import(ImportTaskRequest) returns(common.Status) {}

  rpc ResendSegmentStats(ResendSegmentStatsRequest) returns(ResendSegmentStatsResponse) {}

  rpc AddImportSegment(AddImportSegmentRequest) returns(AddImportSegmentResponse) {}
}

message FlushRequest {
  common.MsgBase base = 1;
  int64 dbID = 2;
  repeated int64 segmentIDs = 3;
  int64 collectionID = 4;
}

message FlushResponse {
  common.Status status = 1;
  int64 dbID = 2;
  int64 collectionID = 3;
  repeated int64 segmentIDs = 4; // newly sealed segments
  repeated int64 flushSegmentIDs = 5; // old flushed segment
  int64 timeOfSeal = 6;
}

message SegmentIDRequest {
  uint32 count = 1;
  string channel_name = 2;
  int64 collectionID = 3;
  int64 partitionID = 4;
  bool isImport = 5;        // Indicate whether this request comes from a bulk insert task.
  int64 importTaskID = 6;   // Needed for segment lock.
}

message AssignSegmentIDRequest {
  int64 nodeID = 1;
  string peer_role = 2;
  repeated SegmentIDRequest segmentIDRequests = 3;
}

message SegmentIDAssignment {
  int64 segID = 1;
  string channel_name = 2;
  uint32 count = 3;
  int64 collectionID = 4;
  int64 partitionID = 5;
  uint64 expire_time = 6;
  common.Status status = 7;
}

message AssignSegmentIDResponse {
  repeated SegmentIDAssignment segIDAssignments = 1;
  common.Status status = 2;
}

message GetSegmentStatesRequest {
  common.MsgBase base = 1;
  repeated int64 segmentIDs = 2;
}

message SegmentStateInfo {
  int64 segmentID = 1;
  common.SegmentState state = 2;
  internal.MsgPosition start_position = 3;
  internal.MsgPosition end_position = 4;
  common.Status status = 5;
}

message GetSegmentStatesResponse {
  common.Status status = 1;
  repeated SegmentStateInfo states = 2;
}

message GetSegmentInfoRequest {
  common.MsgBase base = 1;
  repeated int64 segmentIDs = 2;
  bool includeUnHealthy =3;
}

message GetSegmentInfoResponse {
  common.Status status = 1;
  repeated SegmentInfo infos = 2;
}

message GetInsertBinlogPathsRequest {
  common.MsgBase base = 1;
  int64 segmentID = 2;
}

message GetInsertBinlogPathsResponse {
  repeated int64 fieldIDs = 1;
  repeated internal.StringList paths = 2;
  common.Status status = 3;
}

message GetCollectionStatisticsRequest {
  common.MsgBase base = 1;
  int64 dbID = 2;
  int64 collectionID = 3;
}

message GetCollectionStatisticsResponse {
  repeated common.KeyValuePair stats = 1;
  common.Status status = 2;
}

message GetPartitionStatisticsRequest{
  common.MsgBase base = 1;
  int64 dbID = 2;
  int64 collectionID = 3;
  repeated int64 partitionIDs = 4;
}

message GetPartitionStatisticsResponse {
  repeated common.KeyValuePair stats = 1;
  common.Status status = 2;
}

message GetSegmentInfoChannelRequest {
}

message AcquireSegmentLockRequest {
  common.MsgBase base = 1;
  int64 nodeID = 2;
  repeated int64 segmentIDs = 3;
  int64 taskID = 4;
}

message ReleaseSegmentLockRequest {
  common.MsgBase base = 1;
  int64 nodeID = 2;
  repeated int64 segmentIDs = 3;
  int64 taskID = 4;
}

message VchannelInfo {
  int64 collectionID = 1;
  string channelName = 2;
  internal.MsgPosition seek_position = 3;
  repeated SegmentInfo unflushedSegments = 4; // deprecated, keep it for compatibility
  repeated SegmentInfo flushedSegments = 5; // deprecated, keep it for compatibility
  repeated SegmentInfo dropped_segments = 6; // deprecated, keep it for compatibility
  repeated int64 unflushedSegmentIds = 7;
  repeated int64 flushedSegmentIds = 8;
  repeated int64 dropped_segmentIds = 9;
}

message WatchDmChannelsRequest {
  common.MsgBase base = 1;
  repeated VchannelInfo vchannels = 2;
}

message FlushSegmentsRequest {
  common.MsgBase base = 1;
  int64 dbID = 2;
  int64 collectionID = 3;
  repeated int64 segmentIDs = 4; // segments to flush
}

message SegmentMsg{
  common.MsgBase base = 1;
  SegmentInfo segment = 2;
}

message SegmentInfo {
  int64 ID = 1;
  int64 collectionID = 2;
  int64 partitionID = 3;
  string insert_channel = 4;
  int64 num_of_rows = 5;
  common.SegmentState state = 6;
  int64 max_row_num = 7;
  uint64 last_expire_time = 8;
  internal.MsgPosition start_position = 9;
  internal.MsgPosition dml_position = 10;
  // binlogs consist of insert binlogs
  repeated FieldBinlog binlogs = 11;
  repeated FieldBinlog statslogs = 12;
  // deltalogs consists of delete binlogs. FieldID is not used yet since delete is always applied on primary key
  repeated FieldBinlog deltalogs = 13;
  bool createdByCompaction = 14;
  repeated int64 compactionFrom = 15;
  uint64 dropped_at = 16; // timestamp when segment marked drop
  // A flag indicating if:
  // (1) this segment is created by bulk insert, and
  // (2) the bulk insert task that creates this segment has not yet reached `ImportCompleted` state.
  bool is_importing = 17;
  bool is_fake = 18;
}

message SegmentStartPosition {
  internal.MsgPosition start_position = 1;
  int64 segmentID = 2;
}

message SaveBinlogPathsRequest {
  common.MsgBase base = 1;
  int64 segmentID = 2;
  int64 collectionID = 3;
  repeated FieldBinlog field2BinlogPaths = 4;
  repeated CheckPoint checkPoints = 5;
  repeated SegmentStartPosition start_positions = 6;
  bool flushed = 7;
  repeated FieldBinlog field2StatslogPaths = 8;
  repeated FieldBinlog deltalogs = 9;
  bool dropped = 10;
  bool importing = 11;
}

message CheckPoint {
  int64 segmentID = 1;
  internal.MsgPosition position = 2;
  int64 num_of_rows = 3;
}

message DeltaLogInfo {
  uint64 record_entries = 1;
  uint64 timestamp_from = 2;
  uint64 timestamp_to = 3;
  string delta_log_path = 4;
  int64 delta_log_size = 5;
}

message DataNodeTtMsg {
    common.MsgBase base =1;
    string channel_name = 2;
    uint64 timestamp = 3;
    repeated SegmentStats segments_stats = 4;
}

message SegmentStats {
  int64 SegmentID = 1;
  int64 NumRows = 2;
}

enum ChannelWatchState {
  Uncomplete = 0; // deprecated, keep it for compatibility
  Complete = 1; // deprecated, keep it for compatibility
  ToWatch = 2;
  WatchSuccess = 3;
  WatchFailure = 4;
  ToRelease = 5;
  ReleaseSuccess = 6;
  ReleaseFailure = 7;
}

message ChannelStatus {
  string name = 1;
  ChannelWatchState state=2;
  int64 collectionID = 3;
}

message DataNodeInfo {
  string address = 1;
  int64 version  = 2;
  repeated ChannelStatus channels = 3;
}

message SegmentBinlogs {
  int64 segmentID = 1;
  repeated FieldBinlog fieldBinlogs = 2;
  int64 num_of_rows = 3;
  repeated FieldBinlog statslogs = 4;
  repeated FieldBinlog deltalogs = 5;
  string insert_channel = 6;
}

message FieldBinlog{
  int64 fieldID = 1;
  repeated Binlog binlogs = 2;
}

message Binlog {
  int64 entries_num = 1;
  uint64 timestamp_from = 2;
  uint64 timestamp_to = 3;
  // deprecated
  string log_path = 4;
  int64 log_size = 5;
  int64 logID = 6;
}

message GetRecoveryInfoResponse {
  common.Status status = 1;
  repeated VchannelInfo channels = 2;
  repeated SegmentBinlogs binlogs = 3;
}

message GetRecoveryInfoRequest {
  common.MsgBase base = 1;
  int64 collectionID = 2;
  int64 partitionID = 3;
}

message GetSegmentsByStatesRequest {
  common.MsgBase base = 1;
  int64 collectionID = 2;
  int64 partitionID = 3;
  repeated common.SegmentState states = 4;
}

message GetSegmentsByStatesResponse {
  common.Status status = 1;
  repeated int64 segments = 2;
}

message GetFlushedSegmentsRequest {
  common.MsgBase base = 1;
  int64 collectionID = 2;
  int64 partitionID = 3;
  bool includeUnhealthy = 4;
}

message GetFlushedSegmentsResponse {
  common.Status status = 1;
  repeated int64 segments = 2;
}

message SegmentFlushCompletedMsg {
  common.MsgBase base = 1;
  SegmentInfo segment = 2;
}

message ChannelWatchInfo {
    VchannelInfo vchan= 1;
    int64 startTs = 2;
    ChannelWatchState state = 3;
    // the timeout ts, datanode shall do nothing after it
    // if the timeout is not set(= 0), original infinite retry logic shall be applied
    int64 timeoutTs = 4;
    // the schema of the collection to watch, to avoid get schema rpc issues.
    schema.CollectionSchema schema = 5;
}

enum CompactionType {
  UndefinedCompaction = 0;
  reserved 1;
  MergeCompaction = 2;
  MixCompaction = 3;
}

message CompactionStateRequest {
  common.MsgBase base = 1;
}

message SyncSegmentsRequest {
  int64 planID = 1;
  int64 compacted_to = 2;
  int64 num_of_rows = 3;
  repeated int64 compacted_from = 4;
  repeated FieldBinlog stats_logs = 5;
}

message CompactionSegmentBinlogs {
  int64 segmentID = 1;
  repeated FieldBinlog fieldBinlogs = 2;
  repeated FieldBinlog field2StatslogPaths = 3;
  repeated FieldBinlog deltalogs = 4;
  string insert_channel = 5;
}

message CompactionPlan {
  int64 planID = 1;
  repeated CompactionSegmentBinlogs segmentBinlogs = 2;
  uint64 start_time = 3;
  int32 timeout_in_seconds = 4;
  CompactionType type = 5;
  uint64 timetravel = 6;
  string channel = 7;
  int64 collection_ttl = 8;
}

message CompactionResult {
  int64 planID = 1;
  int64 segmentID = 2;
  int64 num_of_rows = 3;
  repeated FieldBinlog insert_logs = 4;
  repeated FieldBinlog field2StatslogPaths = 5;
  repeated FieldBinlog deltalogs = 6;
  string channel = 7;
}

message CompactionStateResult {
  int64 planID = 1;
  common.CompactionState state = 2;
  CompactionResult result = 3;
}

message CompactionStateResponse {
  common.Status status = 1;
  repeated CompactionStateResult results = 2;
}

// Deprecated
message SegmentFieldBinlogMeta {
  int64  fieldID = 1;
  string binlog_path = 2;
}

message WatchChannelsRequest {
  int64 collectionID = 1;
  repeated string channelNames = 2;
  repeated common.KeyDataPair start_positions = 3;
  schema.CollectionSchema schema = 4; 
}

message WatchChannelsResponse {
  common.Status status = 1;
}

message SetSegmentStateRequest {
  common.MsgBase base = 1;
  int64 segment_id = 2;
  common.SegmentState new_state = 3;
}

message SetSegmentStateResponse {
  common.Status status = 1;
}

message DropVirtualChannelRequest {
  common.MsgBase base = 1;
  string channel_name = 2;
  repeated DropVirtualChannelSegment segments = 3;
}

message DropVirtualChannelSegment {
  int64 segmentID = 1;
  int64 collectionID = 2;
  repeated FieldBinlog field2BinlogPaths = 3;
  repeated FieldBinlog field2StatslogPaths = 4;
  repeated FieldBinlog deltalogs = 5;
  internal.MsgPosition startPosition = 6;
  internal.MsgPosition checkPoint = 7;
  int64 numOfRows = 8;
}

message DropVirtualChannelResponse {
  common.Status status = 1;
}

message ImportTask {
  common.Status status = 1;
  int64 collection_id = 2;                   // target collection ID
  int64 partition_id = 3;                    // target partition ID
  repeated string channel_names = 4;         // target channel names of the collection.
  bool row_based = 5;                        // the file is row-based or column-based
  int64 task_id = 6;                         // id of the task
  repeated string files = 7;                 // file paths to be imported
  repeated common.KeyValuePair infos = 8;    // extra information about the task, bucket, etc.
}

message ImportTaskState {
  common.ImportState stateCode = 1;    // Import state code.
  repeated int64 segments = 2;         // Ids of segments created in import task.
  repeated int64 row_ids = 3;          // Row IDs for the newly inserted rows.
  int64 row_count = 4;                 // # of rows added in the import task.
  string error_message = 5;            // Error message for the failed task.
}

message ImportTaskInfo {
  int64 id = 1;                                 // Task ID.
  int64 request_id = 2 [deprecated = true];     // Request ID of the import task.
  int64 datanode_id = 3;                        // ID of DataNode that processes the task.
  int64 collection_id = 4;                      // Collection ID for the import task.
  int64 partition_id = 5;                       // Partition ID for the import task.
  repeated string channel_names = 6;            // Names of channels for the collection.
  string bucket = 7;                            // Bucket for the import task.
  bool row_based = 8;                           // Boolean indicating whether import files are row-based or column-based.
  repeated string files = 9;                    // A list of files to import.
  int64 create_ts = 10;                         // Timestamp when the import task is created.
  ImportTaskState state = 11;                   // State of the import task.
  string collection_name = 12;                  // Collection name for the import task.
  string partition_name = 13;                   // Partition name for the import task.
  repeated common.KeyValuePair infos = 14;      // extra information about the task, bucket, etc.
  int64 start_ts = 15;                          // Timestamp when the import task is sent to datanode to execute.
}

message ImportTaskResponse {
  common.Status status = 1;
  int64 datanode_id = 2;         // which datanode takes this task
}

message ImportTaskRequest {
  common.MsgBase base = 1;
  ImportTask import_task = 2;          // Target import task.
  repeated int64 working_nodes = 3;    // DataNodes that are currently working.
}

message UpdateSegmentStatisticsRequest {
  common.MsgBase base = 1;
  repeated SegmentStats stats = 2;
}

message UpdateChannelCheckpointRequest {
  common.MsgBase base = 1;
  string vChannel = 2;
  internal.MsgPosition position = 3;
}

message ResendSegmentStatsRequest {
  common.MsgBase base = 1;
}

message ResendSegmentStatsResponse {
  common.Status status = 1;
  repeated int64 seg_resent = 2;
}

message AddImportSegmentRequest {
  common.MsgBase base = 1;
  int64 segment_id = 2;
  string channel_name = 3;
  int64 collection_id = 4;
  int64 partition_id = 5;
  int64 row_num = 6;
  repeated FieldBinlog stats_log = 7;
}

message AddImportSegmentResponse {
  common.Status status = 1;
  bytes channel_pos = 2;
}

message SaveImportSegmentRequest {
  common.MsgBase base = 1;
  int64 segment_id = 2;
  string channel_name = 3;
  int64 collection_id = 4;
  int64 partition_id = 5;
  int64 row_num = 6;
  SaveBinlogPathsRequest save_binlog_path_req = 7;
  bytes dml_position_id = 8;
}

message UnsetIsImportingStateRequest {
  common.MsgBase base = 1;
  repeated int64 segment_ids = 2;       // IDs of segments whose `isImport` states need to be unset.
}

message MarkSegmentsDroppedRequest {
  common.MsgBase base = 1;
  repeated int64 segment_ids = 2;       // IDs of segments that needs to be marked as `dropped`.
}

message SegmentReferenceLock {
  int64 taskID = 1;
  int64 nodeID = 2;
  repeated int64 segmentIDs = 3;
}


message AlterCollectionRequest {
  int64  collectionID = 1;
  schema.CollectionSchema schema = 2;
  repeated int64 partitionIDs = 3;
  repeated common.KeyDataPair start_positions = 4;
  repeated common.KeyValuePair properties = 5;
}