enhance:add support for partition stats for major compaction test(mil…

…vus-io#276) Signed-off-by: MrPresent-Han <[email protected]>
MrPresent-Han · Jun 21, 2024 · f5fceab · f5fceab
1 parent c9bcfdb
commit f5fceab
Show file tree

Hide file tree

Showing 6 changed files with 468 additions and 403 deletions.
diff --git a/models/segment.go b/models/segment.go
@@ -34,6 +34,9 @@ type Segment struct {
 	// Semantic version
 	Version string
 
+	//PartitionStats version
+	PartitionStatsVersion int64
+
 	// etcd segment key
 	key string
 
@@ -126,6 +129,7 @@ func NewSegmentFromV2_2(info *datapbv2.SegmentInfo, key string,
 	}
 
 	s.Version = ">=2.2.0"
+	s.PartitionStatsVersion = info.GetPartitionStatsVersion()
 	return s
 }
 

diff --git a/models/segment_state.go b/models/segment_state.go
@@ -45,18 +45,21 @@ const (
 	SegmentLevelLegacy SegmentLevel = 0
 	SegmentLevelL0     SegmentLevel = 1
 	SegmentLevelL1     SegmentLevel = 2
+	SegmentLevelL2     SegmentLevel = 3
 )
 
 var SegmentLevelName = map[int32]string{
 	0: "Legacy",
 	1: "L0",
 	2: "L1",
+	3: "L2",
 }
 
 var SegmentLevelValue = map[string]int32{
 	"Legacy": 0,
 	"L0":     1,
 	"L1":     2,
+	"L2":     3,
 }
 
 func (x SegmentLevel) String() string {

diff --git a/proto/v2.2/data_coord.proto b/proto/v2.2/data_coord.proto
@@ -110,7 +110,7 @@ service DataNode {
   // 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 CompactionV2(CompactionPlan) returns (common.Status) {}
   rpc GetCompactionState(CompactionStateRequest) returns (CompactionStateResponse) {}
   rpc SyncSegments(SyncSegmentsRequest) returns (common.Status) {}
 
@@ -127,6 +127,10 @@ service DataNode {
   rpc QueryPreImport(QueryPreImportRequest) returns(QueryPreImportResponse) {}
   rpc QueryImport(QueryImportRequest) returns(QueryImportResponse) {}
   rpc DropImport(DropImportRequest) returns(common.Status) {}
+
+  rpc QuerySlot(QuerySlotRequest) returns(QuerySlotResponse) {}
+
+  rpc DropCompactionPlan(DropCompactionPlanRequest) returns(common.Status) {}
 }
 
 message FlushRequest {
@@ -517,6 +521,7 @@ message SyncSegmentsRequest {
   string channel_name = 6;
   int64 partition_id = 7;
   int64 collection_id = 8;
+  map<int64, SyncSegmentInfo> segment_infos = 9;
 }
 
 message CompactionSegmentBinlogs {

diff --git a/proto/v2.2/query_coord.proto b/proto/v2.2/query_coord.proto
@@ -589,6 +589,7 @@ message SealedSegmentsChangeInfo {
 message GetDataDistributionRequest {
     common.MsgBase base = 1;
     map<string, msg.MsgPosition> checkpoints = 2;
+    int64 lastUpdateTs = 3;
 }
 
 message GetDataDistributionResponse {
@@ -597,6 +598,7 @@ message GetDataDistributionResponse {
     repeated SegmentVersionInfo segments = 3;
     repeated ChannelVersionInfo channels = 4;
     repeated LeaderView leader_views = 5;
+    int64 lastModifyTs = 6;
 }
 
 message LeaderView {
@@ -607,6 +609,7 @@ message LeaderView {
     map<int64, msg.MsgPosition> growing_segments = 5;
     int64 TargetVersion = 6;
     int64 num_of_growing_rows = 7;
+    map<int64, int64> partition_stats_versions = 8;
 }
 
 message SegmentDist {
@@ -677,6 +680,7 @@ enum SyncType {
     Set = 1;
     Amend = 2;
     UpdateVersion = 3;
+    UpdatePartitionStats = 4;
 }
 
 message SyncAction {
@@ -691,6 +695,7 @@ message SyncAction {
     int64 TargetVersion = 9;
     repeated int64 droppedInTarget = 10;
     msg.MsgPosition checkpoint = 11;
+    map<int64, int64> partition_stats_versions = 12;
 }
 
 message SyncDistributionRequest {