fix potential bug

wayblink · May 28, 2024 · c851363 · c851363
1 parent 2d69006
commit c851363
Show file tree

Hide file tree

Showing 6 changed files with 58 additions and 27 deletions.
diff --git a/internal/datacoord/compaction.go b/internal/datacoord/compaction.go
@@ -273,9 +273,12 @@ func (c *compactionPlanHandler) enqueueCompaction(task CompactionTask) error {
 	// todo optimize @wayblink
 	// currently compaction trigger v1 v2 both exist and have no shared lock between them to protect
 	// temporarily refuse the enqueue
-	succeed := c.checkAndSetSegmentsCompacting(task)
+	exist, succeed := c.checkAndSetSegmentsCompacting(task)
+	if !exist {
+		return merr.WrapErrIllegalCompactionPlan("segment not exist")
+	}
 	if !succeed {
-		return errors.New("compaction plan conflict")
+		return merr.WrapErrCompactionPlanConflict("segment is compacting")
 	}
 
 	_, span := otel.Tracer(typeutil.DataCoordRole).Start(context.Background(), fmt.Sprintf("Compaction-%s", task.GetType()))
@@ -331,7 +334,7 @@ func (c *compactionPlanHandler) setSegmentsCompacting(task CompactionTask, compa
 	}
 }
 
-func (c *compactionPlanHandler) checkAndSetSegmentsCompacting(task CompactionTask) bool {
+func (c *compactionPlanHandler) checkAndSetSegmentsCompacting(task CompactionTask) (bool, bool) {
 	return c.meta.CheckAndSetSegmentsCompacting(task.GetInputSegments())
 }
 

diff --git a/internal/datacoord/meta.go b/internal/datacoord/meta.go
@@ -53,7 +53,7 @@ type CompactionMeta interface {
 	GetHealthySegment(segID UniqueID) *SegmentInfo
 	UpdateSegmentsInfo(operators ...UpdateOperator) error
 	SetSegmentCompacting(segmentID int64, compacting bool)
-	CheckAndSetSegmentsCompacting(segmentIDs []int64) bool
+	CheckAndSetSegmentsCompacting(segmentIDs []int64) (bool, bool)
 
 	SaveClusteringCompactionTask(task *datapb.CompactionTask) error
 	DropClusteringCompactionTask(task *datapb.CompactionTask) error
@@ -1252,19 +1252,24 @@ func (m *meta) SetSegmentCompacting(segmentID UniqueID, compacting bool) {
 // CheckAndSetSegmentsCompacting check all segments are not compacting
 // if true, set them compacting and return true
 // if false, skip setting and
-func (m *meta) CheckAndSetSegmentsCompacting(segmentIDs []UniqueID) (hasCompactingSegment bool) {
+func (m *meta) CheckAndSetSegmentsCompacting(segmentIDs []UniqueID) (exist, hasCompactingSegment bool) {
 	m.Lock()
 	defer m.Unlock()
 	for _, segmentID := range segmentIDs {
-		hasCompactingSegment = m.segments.GetSegment(segmentID).isCompacting
+		seg := m.segments.GetSegment(segmentID)
+		if seg != nil {
+			hasCompactingSegment = seg.isCompacting
+		} else {
+			return false, false
+		}
 	}
 	if hasCompactingSegment {
-		return false
+		return true, false
 	}
 	for _, segmentID := range segmentIDs {
 		m.segments.SetIsCompacting(segmentID, true)
 	}
-	return true
+	return true, true
 }
 
 // SetSegmentLevel sets level for segment

diff --git a/internal/datacoord/mock_compaction_meta.go b/internal/datacoord/mock_compaction_meta.go
diff --git a/internal/datanode/compaction/clustering_compactor.go b/internal/datanode/compaction/clustering_compactor.go
@@ -549,15 +549,17 @@ func (t *clusteringCompactionTask) mappingSegment(
 			remained++
 
 			currentSize := t.totalBufferSize.Load()
-			// trigger spill
-			if clusterBuffer.writer.GetRowNum() > t.plan.GetMaxSegmentRows() || clusterBuffer.writer.IsFull() {
-				// reach segment/binlog max size
-				t.spillChan <- SpillSignal{
-					buffer: clusterBuffer,
+			if (remained+1)%20 == 0 {
+				// trigger spill
+				if clusterBuffer.writer.GetRowNum() > t.plan.GetMaxSegmentRows() || clusterBuffer.writer.IsFull() {
+					// reach segment/binlog max size
+					t.spillChan <- SpillSignal{
+						buffer: clusterBuffer,
+					}
+				} else if currentSize >= t.getMemoryBufferMiddleWatermark() {
+					// reach spill trigger threshold
+					t.spillChan <- SpillSignal{}
 				}
-			} else if currentSize >= t.getMemoryBufferMiddleWatermark() {
-				// reach spill trigger threshold
-				t.spillChan <- SpillSignal{}
 			}
 
 			// if the total buffer size is too large, block here, wait for memory release by spill
@@ -751,6 +753,7 @@ func (t *clusteringCompactionTask) packBufferToSegment(ctx context.Context, buff
 }
 
 func (t *clusteringCompactionTask) spill(ctx context.Context, buffer *ClusterBuffer) error {
+	log := log.With(zap.Int("bufferID", buffer.id), zap.Int64("bufferSize", buffer.bufferRowNum.Load()))
 	if buffer.writer.IsEmpty() {
 		return nil
 	}
@@ -780,6 +783,7 @@ func (t *clusteringCompactionTask) spill(ctx context.Context, buffer *ClusterBuf
 	buffer.bufferSize.Store(0)
 	buffer.bufferRowNum.Store(0)
 
+	log.Info("finish spill binlogs")
 	if buffer.flushedRowNum > t.plan.GetMaxSegmentRows() {
 		if err := t.packBufferToSegment(ctx, buffer); err != nil {
 			return err

diff --git a/pkg/util/merr/errors.go b/pkg/util/merr/errors.go
@@ -174,12 +174,13 @@ var (
 	// Compaction
 	ErrCompactionReadDeltaLogErr                  = newMilvusError("fail to read delta log", 2300, false)
 	ErrIllegalCompactionPlan                      = newMilvusError("compaction plan illegal", 2301, false)
-	ErrClusteringCompactionClusterNotSupport      = newMilvusError("milvus cluster not support clustering compaction", 2302, false)
-	ErrClusteringCompactionCollectionNotSupport   = newMilvusError("collection not support clustering compaction", 2303, false)
-	ErrClusteringCompactionCollectionIsCompacting = newMilvusError("collection is compacting", 2304, false)
-	ErrClusteringCompactionNotSupportVector       = newMilvusError("vector field clustering compaction is not supported", 2305, false)
-	ErrClusteringCompactionSubmitTaskFail         = newMilvusError("fail to submit task", 2306, true)
-	ErrClusteringCompactionMetaError              = newMilvusError("fail to update meta in clustering compaction", 2307, true)
+	ErrCompactionPlanConflict                     = newMilvusError("compaction plan conflict", 2302, false)
+	ErrClusteringCompactionClusterNotSupport      = newMilvusError("milvus cluster not support clustering compaction", 2303, false)
+	ErrClusteringCompactionCollectionNotSupport   = newMilvusError("collection not support clustering compaction", 2304, false)
+	ErrClusteringCompactionCollectionIsCompacting = newMilvusError("collection is compacting", 2305, false)
+	ErrClusteringCompactionNotSupportVector       = newMilvusError("vector field clustering compaction is not supported", 2306, false)
+	ErrClusteringCompactionSubmitTaskFail         = newMilvusError("fail to submit task", 2307, true)
+	ErrClusteringCompactionMetaError              = newMilvusError("fail to update meta in clustering compaction", 2308, true)
 
 	// General
 	ErrOperationNotSupported = newMilvusError("unsupported operation", 3000, false)

diff --git a/pkg/util/merr/utils.go b/pkg/util/merr/utils.go
@@ -1056,6 +1056,14 @@ func WrapErrIllegalCompactionPlan(msg ...string) error {
 	return err
 }
 
+func WrapErrCompactionPlanConflict(msg ...string) error {
+	err := error(ErrCompactionPlanConflict)
+	if len(msg) > 0 {
+		err = errors.Wrap(err, strings.Join(msg, "->"))
+	}
+	return err
+}
+
 func WrapErrClusteringCompactionClusterNotSupport(msg ...string) error {
 	err := error(ErrClusteringCompactionClusterNotSupport)
 	if len(msg) > 0 {