enhance: Write buffer time range when syncing logs (#28970)

Related to #27675
The timestamp from, to field is not field for new implementation of
writebuffer & sync manager
This pr fills these field for better log information

Signed-off-by: Congqi Xia <congqi.xia@zilliz.com>

internal/datanode/syncmgr/task.go

@@ -188,6 +188,8 @@ func (t *SyncTask) serializeDeleteData() error {
 	t.segmentData[blobPath] = value
 	data.LogSize = int64(len(blob.Value))
 	data.LogPath = blobPath
+	data.TimestampFrom = t.tsFrom
+	data.TimestampTo = t.tsTo
 	data.EntriesNum = t.deleteData.RowCount
 	t.appendDeltalog(data)
 

internal/datanode/writebuffer/insert_buffer.go

@@ -68,6 +68,13 @@ func (b *BufferBase) MinTimestamp() typeutil.Timestamp {
 	return b.startPos.GetTimestamp()
 }
 
+func (b *BufferBase) GetTimeRange() *TimeRange {
+	return &TimeRange{
+		timestampMin: b.TimestampFrom,
+		timestampMax: b.TimestampTo,
+	}
+}
+
 type InsertBuffer struct {
 	BufferBase
 	collSchema *schemapb.CollectionSchema

internal/datanode/writebuffer/segment_buffer.go

@@ -1,6 +1,8 @@
 package writebuffer
 
 import (
+	"math"
+
 	"github.com/milvus-io/milvus-proto/go-api/v2/msgpb"
 	"github.com/milvus-io/milvus-proto/go-api/v2/schemapb"
 	"github.com/milvus-io/milvus/internal/storage"
@@ -48,12 +50,36 @@ func (buf *segmentBuffer) EarliestPosition() *msgpb.MsgPosition {
 	return getEarliestCheckpoint(buf.insertBuffer.startPos, buf.deltaBuffer.startPos)
 }
 
+func (buf *segmentBuffer) GetTimeRange() *TimeRange {
+	result := &TimeRange{
+		timestampMin: math.MaxUint64,
+		timestampMax: 0,
+	}
+	if buf.insertBuffer != nil {
+		result.Merge(buf.insertBuffer.GetTimeRange())
+	}
+	if buf.deltaBuffer != nil {
+		result.Merge(buf.deltaBuffer.GetTimeRange())
+	}
+
+	return result
+}
+
 // TimeRange is a range of timestamp contains the min-timestamp and max-timestamp
 type TimeRange struct {
 	timestampMin typeutil.Timestamp
 	timestampMax typeutil.Timestamp
 }
 
+func (tr *TimeRange) Merge(other *TimeRange) {
+	if other.timestampMin < tr.timestampMin {
+		tr.timestampMin = other.timestampMin
+	}
+	if other.timestampMax > tr.timestampMax {
+		tr.timestampMax = other.timestampMax
+	}
+}
+
 func getEarliestCheckpoint(cps ...*msgpb.MsgPosition) *msgpb.MsgPosition {
 	var result *msgpb.MsgPosition
 	for _, cp := range cps {

internal/datanode/writebuffer/write_buffer.go

@@ -236,18 +236,19 @@ func (wb *writeBufferBase) getOrCreateBuffer(segmentID int64) *segmentBuffer {
 	return buffer
 }
 
-func (wb *writeBufferBase) yieldBuffer(segmentID int64) (*storage.InsertData, *storage.DeleteData, *msgpb.MsgPosition) {
+func (wb *writeBufferBase) yieldBuffer(segmentID int64) (*storage.InsertData, *storage.DeleteData, *TimeRange, *msgpb.MsgPosition) {
 	buffer, ok := wb.buffers[segmentID]
 	if !ok {
-		return nil, nil, nil
+		return nil, nil, nil, nil
 	}
 
 	// remove buffer and move it to sync manager
 	delete(wb.buffers, segmentID)
 	start := buffer.EarliestPosition()
+	timeRange := buffer.GetTimeRange()
 	insert, delta := buffer.Yield()
 
-	return insert, delta, start
+	return insert, delta, timeRange, start
 }
 
 // bufferInsert transform InsertMsg into bufferred InsertData and returns primary key field data for future usage.
@@ -328,14 +329,21 @@ func SpaceCreatorFunc(segmentID int64, collSchema *schemapb.CollectionSchema, ar
 }
 
 func (wb *writeBufferBase) getSyncTask(ctx context.Context, segmentID int64) syncmgr.Task {
+	log := log.Ctx(ctx).With(
+		zap.Int64("segmentID", segmentID),
+	)
 	segmentInfo, ok := wb.metaCache.GetSegmentByID(segmentID) // wb.metaCache.GetSegmentsBy(metacache.WithSegmentIDs(segmentID))
 	if !ok {
-		log.Ctx(ctx).Warn("segment info not found in meta cache", zap.Int64("segmentID", segmentID))
+		log.Warn("segment info not found in meta cache", zap.Int64("segmentID", segmentID))
 		return nil
 	}
 	var batchSize int64
+	var tsFrom, tsTo uint64
 
-	insert, delta, startPos := wb.yieldBuffer(segmentID)
+	insert, delta, timeRange, startPos := wb.yieldBuffer(segmentID)
+	if timeRange != nil {
+		tsFrom, tsTo = timeRange.timestampMin, timeRange.timestampMax
+	}
 
 	actions := []metacache.SegmentAction{metacache.RollStats()}
 	if insert != nil {
@@ -361,6 +369,7 @@ func (wb *writeBufferBase) getSyncTask(ctx context.Context, segmentID int64) syn
 			WithChannelName(wb.channelName).
 			WithSegmentID(segmentID).
 			WithStartPosition(startPos).
+			WithTimeRange(tsFrom, tsTo).
 			WithLevel(segmentInfo.Level()).
 			WithCheckpoint(wb.checkpoint).
 			WithSchema(wb.collSchema).
@@ -386,6 +395,7 @@ func (wb *writeBufferBase) getSyncTask(ctx context.Context, segmentID int64) syn
 			WithChannelName(wb.channelName).
 			WithSegmentID(segmentID).
 			WithStartPosition(startPos).
+			WithTimeRange(tsFrom, tsTo).
 			WithLevel(segmentInfo.Level()).
 			WithCheckpoint(wb.checkpoint).
 			WithSchema(wb.collSchema).