Repository: milvus-io/milvus
File: internal/datanode/l0_compactor.go (Go)

Code before:
		zap.Int64("planID", t.plan.GetPlanID()),
		zap.String("type", t.plan.GetType().String()),
		zap.Int("target segment counts", len(l1Segments)),
	)

	// just for logging
	l1SegmentIDs := lo.Map(l1Segments, func(segment *datapb.CompactionSegmentBinlogs, _ int) int64 {
		return segment.GetSegmentID()
	})

	var (
		resultSegments  = make(map[int64]*datapb.CompactionSegment)
		alteredSegments = make(map[int64]*storage.DeleteData)
	)

	segmentBFs, err := t.loadBF(l1Segments)
	if err != nil {
		return nil, err
	}
	for segID, deltaLogs := range totalDeltalogs {
		log := log.With(zap.Int64("levelzero segment", segID))

		log.Info("Linear L0 compaction start processing segment")
		allIters, err := t.loadDelta(ctx, deltaLogs)
		if err != nil {
			log.Warn("Linear L0 compaction loadDelta fail", zap.Int64s("target segments", l1SegmentIDs), zap.Error(err))
			return nil, err
		}

		t.splitDelta(ctx, allIters, alteredSegments, segmentBFs)

		err = t.uploadByCheck(ctx, true, alteredSegments, resultSegments)
		if err != nil {
			log.Warn("Linear L0 compaction upload buffer fail", zap.Int64s("target segments", l1SegmentIDs), zap.Error(err))
			return nil, err
		}
	}

	err = t.uploadByCheck(ctx, false, alteredSegments, resultSegments)
	if err != nil {
		log.Warn("Linear L0 compaction upload all buffer fail", zap.Int64s("target segment", l1SegmentIDs), zap.Error(err))
		return nil, err
	}
	log.Info("Linear L0 compaction finished", zap.Duration("elapse", t.tr.RecordSpan()))
	return lo.Values(resultSegments), nil
}

func (t *levelZeroCompactionTask) batchProcess(ctx context.Context, l1Segments []*datapb.CompactionSegmentBinlogs, deltaLogs ...[]string) ([]*datapb.CompactionSegment, error) {
	log := log.Ctx(t.ctx).With(
		zap.Int64("planID", t.plan.GetPlanID()),
		zap.String("type", t.plan.GetType().String()),

Review comment (suggestion):
actually, L0 compaction will target to L1 &L2, any segments not L0, so `L1SegmentIDs` are misleading here. Please change back.

Code after:
		zap.Int64("planID", t.plan.GetPlanID()),
		zap.String("type", t.plan.GetType().String()),
		zap.Int("target segment counts", len(targetSegments)),
	)

	// just for logging
	targetSegmentIDs := lo.Map(targetSegments, func(segment *datapb.CompactionSegmentBinlogs, _ int) int64 {
		return segment.GetSegmentID()
	})

	var (
		resultSegments  = make(map[int64]*datapb.CompactionSegment)
		alteredSegments = make(map[int64]*storage.DeleteData)
	)

	segmentBFs, err := t.loadBF(targetSegments)
	if err != nil {
		return nil, err
	}
	for segID, deltaLogs := range totalDeltalogs {
		log := log.With(zap.Int64("levelzero segment", segID))

		log.Info("Linear L0 compaction start processing segment")
		allIters, err := t.loadDelta(ctx, deltaLogs)
		if err != nil {
			log.Warn("Linear L0 compaction loadDelta fail", zap.Int64s("target segments", targetSegmentIDs), zap.Error(err))
			return nil, err
		}

		t.splitDelta(ctx, allIters, alteredSegments, segmentBFs)

		err = t.uploadByCheck(ctx, true, alteredSegments, resultSegments)
		if err != nil {
			log.Warn("Linear L0 compaction upload buffer fail", zap.Int64s("target segments", targetSegmentIDs), zap.Error(err))
			return nil, err
		}
	}

	err = t.uploadByCheck(ctx, false, alteredSegments, resultSegments)
	if err != nil {
		log.Warn("Linear L0 compaction upload all buffer fail", zap.Int64s("target segment", targetSegmentIDs), zap.Error(err))
		return nil, err
	}
	log.Info("Linear L0 compaction finished", zap.Duration("elapse", t.tr.RecordSpan()))
	return lo.Values(resultSegments), nil
}

func (t *levelZeroCompactionTask) batchProcess(ctx context.Context, targetSegments []*datapb.CompactionSegmentBinlogs, deltaLogs ...[]string) ([]*datapb.CompactionSegment, error) {
	log := log.Ctx(t.ctx).With(
		zap.Int64("planID", t.plan.GetPlanID()),
		zap.String("type", t.plan.GetType().String()),