Fix ut

Signed-off-by: yangxuan <[email protected]>
XuanYang-cn · May 8, 2024 · 901951d · 901951d
1 parent 321ee79
commit 901951d
Show file tree

Hide file tree

Showing 2 changed files with 22 additions and 11 deletions.
diff --git a/internal/datanode/l0_compactor.go b/internal/datanode/l0_compactor.go
@@ -341,6 +341,7 @@ func (t *levelZeroCompactionTask) composeDeltalog(segmentID int64, dData *storag
 
 func (t *levelZeroCompactionTask) uploadByCheck(ctx context.Context, requireCheck bool, alteredSegments map[int64]*storage.DeleteData, resultSegments map[int64]*datapb.CompactionSegment) error {
 	allBlobs := make(map[string][]byte)
+	tmpResults := make(map[int64]*datapb.CompactionSegment)
 	for segID, dData := range alteredSegments {
 		if !requireCheck || (dData.Size() >= paramtable.Get().DataNodeCfg.FlushDeleteBufferBytes.GetAsInt64()) {
 			blobs, binlog, err := t.composeDeltalog(segID, dData)
@@ -349,22 +350,32 @@ func (t *levelZeroCompactionTask) uploadByCheck(ctx context.Context, requireChec
 				return err
 			}
 			allBlobs = lo.Assign(blobs, allBlobs)
-			if _, ok := resultSegments[segID]; !ok {
-				resultSegments[segID] = &datapb.CompactionSegment{
-					SegmentID: segID,
-					Deltalogs: []*datapb.FieldBinlog{{Binlogs: []*datapb.Binlog{binlog}}},
-					Channel:   t.plan.GetChannel(),
-				}
-			} else {
-				resultSegments[segID].Deltalogs[0].Binlogs = append(resultSegments[segID].Deltalogs[0].Binlogs, binlog)
+			tmpResults[segID] = &datapb.CompactionSegment{
+				SegmentID: segID,
+				Deltalogs: []*datapb.FieldBinlog{{Binlogs: []*datapb.Binlog{binlog}}},
+				Channel:   t.plan.GetChannel(),
 			}
-
 			delete(alteredSegments, segID)
 		}
 	}
+
+	if len(allBlobs) == 0 {
+		return nil
+	}
+
 	if err := t.Upload(ctx, allBlobs); err != nil {
 		log.Warn("L0 compaction upload blobs fail", zap.Error(err))
 		return err
 	}
+
+	for segID, compSeg := range tmpResults {
+		if _, ok := resultSegments[segID]; !ok {
+			resultSegments[segID] = compSeg
+		} else {
+			binlog := compSeg.Deltalogs[0].Binlogs[0]
+			resultSegments[segID].Deltalogs[0].Binlogs = append(resultSegments[segID].Deltalogs[0].Binlogs, binlog)
+		}
+	}
+
 	return nil
 }
diff --git a/internal/datanode/l0_compactor_test.go b/internal/datanode/l0_compactor_test.go
@@ -220,7 +220,7 @@ func (s *LevelZeroCompactionTaskSuite) TestCompactLinear() {
 		RunAndReturn(func(paths ...string) string {
 			return path.Join(paths...)
 		}).Times(2)
-	s.mockBinlogIO.EXPECT().Upload(mock.Anything, mock.Anything).Return(nil).Times(2)
+	s.mockBinlogIO.EXPECT().Upload(mock.Anything, mock.Anything).Return(nil).Once()
 
 	s.Require().Equal(plan.GetPlanID(), s.task.getPlanID())
 	s.Require().Equal(plan.GetChannel(), s.task.getChannelName())
@@ -322,7 +322,7 @@ func (s *LevelZeroCompactionTaskSuite) TestCompactBatch() {
 		RunAndReturn(func(paths ...string) string {
 			return path.Join(paths...)
 		}).Times(2)
-	s.mockBinlogIO.EXPECT().Upload(mock.Anything, mock.Anything).Return(nil).Times(2)
+	s.mockBinlogIO.EXPECT().Upload(mock.Anything, mock.Anything).Return(nil).Once()
 
 	l0Segments := lo.Filter(s.task.plan.GetSegmentBinlogs(), func(s *datapb.CompactionSegmentBinlogs, _ int) bool {
 		return s.Level == datapb.SegmentLevel_L0