fix: [Cherry-pick]Pick major compaction fixs and optimizations

configs/milvus.yaml

@@ -649,8 +649,10 @@ common:
   traceLogMode: 0 # trace request info
   bloomFilterSize: 100000 # bloom filter initial size
   maxBloomFalsePositive: 0.001 # max false positive rate for bloom filter
+  # clustering key/compaction related
   usePartitionKeyAsClusteringKey: false
   useVectorAsClusteringKey: false
+  enableVectorClusteringKey: false
 
 # QuotaConfig, configurations of Milvus quota and limits.
 # By default, we enable:

internal/datacoord/compaction_task_clustering.go

@@ -177,6 +177,8 @@ func (t *clusteringCompactionTask) BuildCompactionRequest() (*datapb.CompactionP
 
 func (t *clusteringCompactionTask) processPipelining() error {
 	log := log.With(zap.Int64("triggerID", t.TriggerID), zap.Int64("collectionID", t.GetCollectionID()), zap.Int64("planID", t.GetPlanID()))
+	ts := time.Now().UnixMilli()
+	t.updateAndSaveTaskMeta(setStartTime(ts))
 	var operators []UpdateOperator
 	for _, segID := range t.InputSegments {
 		operators = append(operators, UpdateSegmentLevelOperator(segID, datapb.SegmentLevel_L2))

internal/datacoord/handler.go

@@ -134,6 +134,7 @@
 			zap.String("channel", channel.GetName()),
 			zap.Int("numOfSegments", len(segments)),
 			zap.Int("indexed segment", len(indexedSegments)),
+			zap.Int64("currentPartitionStatsVersion", currentPartitionStatsVersion),
 		)
 		unIndexedIDs := make(typeutil.UniqueSet)
 
@@ -145,13 +146,23 @@
 				// Skip bulk insert segments.
 				continue
 			}
-			if s.GetLevel() == datapb.SegmentLevel_L2 && s.PartitionStatsVersion > currentPartitionStatsVersion {
-				// skip major compaction not fully completed.
+			if s.GetLevel() == datapb.SegmentLevel_L2 && s.PartitionStatsVersion != currentPartitionStatsVersion {
+				// in the process of L2 compaction, newly generated segment may be visible before the whole L2 compaction Plan
+				// is finished, we have to skip these fast-finished segment because all segments in one L2 Batch must be
+				// seen atomically, otherwise users will see intermediate result
 				continue
 			}
 			segmentInfos[s.GetID()] = s
 			switch {
 			case s.GetState() == commonpb.SegmentState_Dropped:
+				if s.GetLevel() == datapb.SegmentLevel_L2 && s.GetPartitionStatsVersion() == currentPartitionStatsVersion {
+					// if segment.partStatsVersion is equal to currentPartitionStatsVersion,
+					// it must have been indexed, this is guaranteed by clustering compaction process
+					// this is to ensure that the current valid L2 compaction produce is available to search/query
+					// to avoid insufficient data
+					indexedIDs.Insert(s.GetID())
+					continue
+				}
 				droppedIDs.Insert(s.GetID())
 			case !isFlushState(s.GetState()):
 				growingIDs.Insert(s.GetID())