fix: Use actual data timestamps for imported segment positions (#47276)

This PR:
1. Modifies the import mechanism to use actual timestamps from the
imported data for segment positions instead of using the channel
checkpoint.
2. Fix writeDelta() to extract actual min/max timestamps from
deltaData.Tss instead of using pack.tsFrom/tsTo.
3. Integration test TestBinlogImport updated to verify L1 and L0 segment
positions.

issue: https://github.com/milvus-io/milvus/issues/47275

---------

Signed-off-by: bigsheeper <yihao.dai@zilliz.com>

internal/datacoord/import_checker.go

@@ -442,7 +442,7 @@ func (c *importChecker) checkIndexBuildingJob(job ImportJob) {
 	metrics.ImportJobLatency.WithLabelValues(metrics.ImportStageWaitL0Import).Observe(float64(buildIndexDuration.Milliseconds()))
 	log.Info("import job l0 import done", zap.Duration("jobTimeCost/l0Import", waitL0ImportDuration))
 
-	if c.updateSegmentState(originSegmentIDs, statsSegmentIDs) {
+	if c.unsetSegmentImporting(originSegmentIDs, statsSegmentIDs) {
 		return
 	}
 	// all finished, update import job state to `Completed`.
@@ -475,7 +475,8 @@ func (c *importChecker) waitL0ImortTaskDone(job ImportJob) bool {
 	return false
 }
 
-func (c *importChecker) updateSegmentState(originSegmentIDs, statsSegmentIDs []int64) bool {
+// unsetSegmentImporting unsets the isImporting flag for segments.
+func (c *importChecker) unsetSegmentImporting(originSegmentIDs, statsSegmentIDs []int64) bool {
 	// Here, all segment indexes have been successfully built, try unset isImporting flag for all segments.
 	isImportingSegments := lo.Filter(append(originSegmentIDs, statsSegmentIDs...), func(segmentID int64, _ int) bool {
 		segment := c.meta.GetSegment(c.ctx, segmentID)
@@ -485,21 +486,10 @@ func (c *importChecker) updateSegmentState(originSegmentIDs, statsSegmentIDs []i
 		}
 		return segment.GetIsImporting()
 	})
-	channels, err := c.meta.GetSegmentsChannels(isImportingSegments)
-	if err != nil {
-		log.Warn("get segments channels failed", zap.Error(err))
-		return true
-	}
+
 	for _, segmentID := range isImportingSegments {
-		channelCP := c.meta.GetChannelCheckpoint(channels[segmentID])
-		if channelCP == nil {
-			log.Warn("nil channel checkpoint")
-			return true
-		}
-		op1 := UpdateStartPosition([]*datapb.SegmentStartPosition{{StartPosition: channelCP, SegmentID: segmentID}})
-		op2 := UpdateDmlPosition(segmentID, channelCP)
-		op3 := UpdateIsImporting(segmentID, false)
-		err = c.meta.UpdateSegmentsInfo(c.ctx, op1, op2, op3)
+		op := UpdateIsImporting(segmentID, false)
+		err := c.meta.UpdateSegmentsInfo(c.ctx, op)
 		if err != nil {
 			log.Warn("update import segment failed", zap.Error(err))
 			return true

internal/datacoord/import_task_import.go

@@ -18,6 +18,7 @@ package datacoord
 
 import (
 	"context"
+	"math"
 	"strconv"
 	"time"
 
@@ -29,6 +30,7 @@ import (
 	"github.com/milvus-io/milvus/internal/datacoord/session"
 	"github.com/milvus-io/milvus/internal/json"
 	"github.com/milvus-io/milvus/internal/metastore/kv/binlog"
+	"github.com/milvus-io/milvus/internal/util/importutilv2"
 	"github.com/milvus-io/milvus/pkg/v2/log"
 	"github.com/milvus-io/milvus/pkg/v2/metrics"
 	"github.com/milvus-io/milvus/pkg/v2/proto/datapb"
@@ -206,6 +208,7 @@ func (t *importTask) QueryTaskOnWorker(cluster session.Cluster) {
 	}
 	if resp.GetState() == datapb.ImportTaskStateV2_Completed {
 		totalRows := int64(0)
+		job := t.importMeta.GetJob(context.TODO(), t.GetJobID())
 		for _, info := range resp.GetImportSegmentsInfo() {
 			// try to parse path and fill logID
 			err = binlog.CompressBinLogs(info.GetBinlogs(), info.GetDeltalogs(), info.GetStatslogs(), info.GetBm25Logs())
@@ -214,10 +217,21 @@ func (t *importTask) QueryTaskOnWorker(cluster session.Cluster) {
 					WrapTaskLog(t, zap.Int64("segmentID", info.GetSegmentID()), zap.Error(err))...)
 				return
 			}
-			op1 := UpdateBinlogsOperator(info.GetSegmentID(), info.GetBinlogs(), info.GetStatslogs(), info.GetDeltalogs(), info.GetBm25Logs())
+
+			// Extract actual timestamps from binlogs for segment positions
+			var minTs, maxTs uint64
+			isL0Import := importutilv2.IsL0Import(job.GetOptions())
+			if isL0Import {
+				minTs, maxTs = extractTimestampFromBinlogs(info.GetDeltalogs())
+			} else {
+				minTs, maxTs = extractTimestampFromBinlogs(info.GetBinlogs())
+			}
+
+			opBinlog := UpdateBinlogsOperator(info.GetSegmentID(), info.GetBinlogs(), info.GetStatslogs(), info.GetDeltalogs(), info.GetBm25Logs())
 			opManifest := UpdateManifest(info.GetSegmentID(), info.GetManifestPath())
-			op2 := UpdateStatusOperator(info.GetSegmentID(), commonpb.SegmentState_Flushed)
-			err = t.meta.UpdateSegmentsInfo(context.TODO(), op1, opManifest, op2)
+			opState := UpdateStatusOperator(info.GetSegmentID(), commonpb.SegmentState_Flushed)
+			opPosition := UpdateImportSegmentPosition(info.GetSegmentID(), minTs, maxTs)
+			err = t.meta.UpdateSegmentsInfo(context.TODO(), opBinlog, opManifest, opState, opPosition)
 			if err != nil {
 				updateErr := t.importMeta.UpdateJob(context.TODO(), t.GetJobID(), UpdateJobState(internalpb.ImportJobState_Failed), UpdateJobReason(err.Error()))
 				if updateErr != nil {
@@ -226,7 +240,11 @@ func (t *importTask) QueryTaskOnWorker(cluster session.Cluster) {
 				log.Warn("update import segment binlogs failed", WrapTaskLog(t, zap.String("err", err.Error()))...)
 				return
 			}
-			log.Info("update import segment info done", WrapTaskLog(t, zap.Int64("segmentID", info.GetSegmentID()), zap.Any("segmentInfo", info))...)
+			log.Info("update import segment info done", WrapTaskLog(t,
+				zap.Int64("segmentID", info.GetSegmentID()),
+				zap.Uint64("minTs", minTs),
+				zap.Uint64("maxTs", maxTs),
+				zap.Any("segmentInfo", info))...)
 			totalRows += info.GetImportedRows()
 		}
 		completeTime := time.Now().Format("2006-01-02T15:04:05Z07:00")
@@ -286,3 +304,22 @@ func (t *importTask) MarshalJSON() ([]byte, error) {
 	}
 	return json.Marshal(importTask)
 }
+
+// extractTimestampFromBinlogs extracts min and max timestamps from binlogs.
+// The timestamps are stored in Binlog.TimestampFrom and Binlog.TimestampTo
+// by BulkPackWriterV2.writeInserts() during import sync.
+func extractTimestampFromBinlogs(binlogs []*datapb.FieldBinlog) (minTs, maxTs uint64) {
+	minTs = math.MaxUint64
+	maxTs = 0
+	for _, fieldBinlog := range binlogs {
+		for _, binlog := range fieldBinlog.GetBinlogs() {
+			if binlog.GetTimestampFrom() < minTs {
+				minTs = binlog.GetTimestampFrom()
+			}
+			if binlog.GetTimestampTo() > maxTs {
+				maxTs = binlog.GetTimestampTo()
+			}
+		}
+	}
+	return minTs, maxTs
+}

internal/datacoord/import_task_import_test.go

@@ -18,6 +18,7 @@ package datacoord
 
 import (
 	"context"
+	"math"
 	"testing"
 	"time"
 
@@ -413,10 +414,19 @@ func TestImportTask_QueryTaskOnWorker(t *testing.T) {
 		catalog.EXPECT().ListPreImportTasks(mock.Anything).Return(nil, nil)
 		catalog.EXPECT().ListImportTasks(mock.Anything).Return(nil, nil)
 		catalog.EXPECT().SaveImportTask(mock.Anything, mock.Anything).Return(nil)
+		catalog.EXPECT().SaveImportJob(mock.Anything, mock.Anything).Return(nil)
 
 		im, err := NewImportMeta(context.TODO(), catalog, nil, nil)
 		assert.NoError(t, err)
 
+		var job ImportJob = &importJob{
+			ImportJob: &datapb.ImportJob{
+				JobID: 1,
+			},
+		}
+		err = im.AddJob(context.TODO(), job)
+		assert.NoError(t, err)
+
 		taskProto := &datapb.ImportTaskV2{
 			JobID:        1,
 			TaskID:       2,
@@ -476,6 +486,101 @@ func TestImportTask_QueryTaskOnWorker(t *testing.T) {
 	})
 }
 
+func TestExtractTimestampFromBinlogs(t *testing.T) {
+	t.Run("empty binlogs", func(t *testing.T) {
+		minTs, maxTs := extractTimestampFromBinlogs(nil)
+		assert.Equal(t, uint64(math.MaxUint64), minTs)
+		assert.Equal(t, uint64(0), maxTs)
+	})
+
+	t.Run("empty field binlogs", func(t *testing.T) {
+		binlogs := []*datapb.FieldBinlog{}
+		minTs, maxTs := extractTimestampFromBinlogs(binlogs)
+		assert.Equal(t, uint64(math.MaxUint64), minTs)
+		assert.Equal(t, uint64(0), maxTs)
+	})
+
+	t.Run("single binlog", func(t *testing.T) {
+		binlogs := []*datapb.FieldBinlog{
+			{
+				FieldID: 100,
+				Binlogs: []*datapb.Binlog{
+					{
+						TimestampFrom: 1000,
+						TimestampTo:   2000,
+					},
+				},
+			},
+		}
+		minTs, maxTs := extractTimestampFromBinlogs(binlogs)
+		assert.Equal(t, uint64(1000), minTs)
+		assert.Equal(t, uint64(2000), maxTs)
+	})
+
+	t.Run("multiple binlogs in single field", func(t *testing.T) {
+		binlogs := []*datapb.FieldBinlog{
+			{
+				FieldID: 100,
+				Binlogs: []*datapb.Binlog{
+					{
+						TimestampFrom: 1000,
+						TimestampTo:   2000,
+					},
+					{
+						TimestampFrom: 500,
+						TimestampTo:   3000,
+					},
+				},
+			},
+		}
+		minTs, maxTs := extractTimestampFromBinlogs(binlogs)
+		assert.Equal(t, uint64(500), minTs)
+		assert.Equal(t, uint64(3000), maxTs)
+	})
+
+	t.Run("multiple fields with multiple binlogs", func(t *testing.T) {
+		binlogs := []*datapb.FieldBinlog{
+			{
+				FieldID: 100,
+				Binlogs: []*datapb.Binlog{
+					{
+						TimestampFrom: 1000,
+						TimestampTo:   2000,
+					},
+				},
+			},
+			{
+				FieldID: 101,
+				Binlogs: []*datapb.Binlog{
+					{
+						TimestampFrom: 500,
+						TimestampTo:   1500,
+					},
+					{
+						TimestampFrom: 800,
+						TimestampTo:   3000,
+					},
+				},
+			},
+		}
+		minTs, maxTs := extractTimestampFromBinlogs(binlogs)
+		assert.Equal(t, uint64(500), minTs)
+		assert.Equal(t, uint64(3000), maxTs)
+	})
+
+	t.Run("field binlog with no binlogs inside", func(t *testing.T) {
+		binlogs := []*datapb.FieldBinlog{
+			{
+				FieldID: 100,
+				Binlogs: []*datapb.Binlog{},
+			},
+		}
+		minTs, maxTs := extractTimestampFromBinlogs(binlogs)
+		assert.Equal(t, uint64(math.MaxUint64), minTs)
+		assert.Equal(t, uint64(0), maxTs)
+	})
+}
+
 func TestImportTask_DropTaskOnWorker(t *testing.T) {
 	t.Run("DropImport rpc failed", func(t *testing.T) {
 		catalog := mocks.NewDataCoordCatalog(t)

internal/datacoord/import_util.go

@@ -246,11 +246,6 @@ func AllocImportSegment(ctx context.Context,
 			return nil, err
 		}
 	}
-	position := &msgpb.MsgPosition{
-		ChannelName: channelName,
-		MsgID:       nil,
-		Timestamp:   ts,
-	}
 
 	segmentInfo := &datapb.SegmentInfo{
 		ID:             id,
@@ -262,8 +257,6 @@ func AllocImportSegment(ctx context.Context,
 		MaxRowNum:      0,
 		Level:          level,
 		LastExpireTime: math.MaxUint64,
-		StartPosition:  position,
-		DmlPosition:    position,
 		StorageVersion: storageVersion,
 	}
 	segmentInfo.IsImporting = true

internal/datacoord/meta.go

@@ -1262,6 +1262,35 @@ func UpdateIsImporting(segmentID int64, isImporting bool) UpdateOperator {
 	}
 }
 
+// UpdateImportSegmentPosition updates the segment's StartPosition and DmlPosition
+// for import segments using actual timestamps from the imported data.
+// Unlike UpdateStartPosition/UpdateDmlPosition, this operator allows nil MsgID
+// since import segments don't have message queue positions.
+func UpdateImportSegmentPosition(segmentID int64, minTs, maxTs uint64) UpdateOperator {
+	return func(modPack *updateSegmentPack) bool {
+		segment := modPack.Get(segmentID)
+		if segment == nil {
+			log.Ctx(context.TODO()).Warn("meta update: update import segment position failed - segment not found",
+				zap.Int64("segmentID", segmentID))
+			return false
+		}
+		channelName := segment.GetInsertChannel()
+		// Use actual min timestamp for StartPosition
+		segment.StartPosition = &msgpb.MsgPosition{
+			ChannelName: channelName,
+			MsgID:       nil,
+			Timestamp:   minTs,
+		}
+		// Use actual max timestamp for DmlPosition
+		segment.DmlPosition = &msgpb.MsgPosition{
+			ChannelName: channelName,
+			MsgID:       nil,
+			Timestamp:   maxTs,
+		}
+		return true
+	}
+}
+
 // UpdateAsDroppedIfEmptyWhenFlushing updates segment state to Dropped if segment is empty and in Flushing state
 // It's used to make a empty flushing segment to be dropped directly.
 func UpdateAsDroppedIfEmptyWhenFlushing(segmentID int64) UpdateOperator {

internal/datacoord/segment_operator_test.go

@@ -19,6 +19,7 @@ package datacoord
 import (
 	"testing"
 
+	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/suite"
 
 	"github.com/milvus-io/milvus/pkg/v2/proto/datapb"
@@ -47,3 +48,69 @@ func (s *TestSegmentOperatorSuite) TestSetMaxRowCount() {
 func TestSegmentOperators(t *testing.T) {
 	suite.Run(t, new(TestSegmentOperatorSuite))
 }
+
+func TestUpdateImportSegmentPosition(t *testing.T) {
+	t.Run("segment not found", func(t *testing.T) {
+		// Create a meta with empty segments to properly test the "not found" case
+		segments := NewSegmentsInfo()
+		m := &meta{segments: segments}
+		modPack := &updateSegmentPack{
+			meta:     m,
+			segments: make(map[int64]*SegmentInfo),
+		}
+		op := UpdateImportSegmentPosition(100, 1000, 2000)
+		result := op(modPack)
+		assert.False(t, result)
+	})
+
+	t.Run("update position successfully", func(t *testing.T) {
+		segment := &SegmentInfo{
+			SegmentInfo: &datapb.SegmentInfo{
+				ID:            100,
+				InsertChannel: "test_channel",
+			},
+		}
+		modPack := &updateSegmentPack{
+			segments: map[int64]*SegmentInfo{
+				100: segment,
+			},
+		}
+		op := UpdateImportSegmentPosition(100, 1000, 2000)
+		result := op(modPack)
+		assert.True(t, result)
+
+		// Verify StartPosition
+		assert.NotNil(t, segment.GetStartPosition())
+		assert.Equal(t, "test_channel", segment.GetStartPosition().GetChannelName())
+		assert.Nil(t, segment.GetStartPosition().GetMsgID())
+		assert.Equal(t, uint64(1000), segment.GetStartPosition().GetTimestamp())
+
+		// Verify DmlPosition
+		assert.NotNil(t, segment.GetDmlPosition())
+		assert.Equal(t, "test_channel", segment.GetDmlPosition().GetChannelName())
+		assert.Nil(t, segment.GetDmlPosition().GetMsgID())
+		assert.Equal(t, uint64(2000), segment.GetDmlPosition().GetTimestamp())
+	})
+
+	t.Run("update position with zero timestamps", func(t *testing.T) {
+		segment := &SegmentInfo{
+			SegmentInfo: &datapb.SegmentInfo{
+				ID:            101,
+				InsertChannel: "channel_2",
+			},
+		}
+		modPack := &updateSegmentPack{
+			segments: map[int64]*SegmentInfo{
+				101: segment,
+			},
+		}
+		op := UpdateImportSegmentPosition(101, 0, 0)
+		result := op(modPack)
+		assert.True(t, result)
+
+		assert.NotNil(t, segment.GetStartPosition())
+		assert.Equal(t, uint64(0), segment.GetStartPosition().GetTimestamp())
+		assert.NotNil(t, segment.GetDmlPosition())
+		assert.Equal(t, uint64(0), segment.GetDmlPosition().GetTimestamp())
+	})
+}

internal/datanode/importv2/util.go

@@ -102,7 +102,6 @@ func NewSyncTask(ctx context.Context,
 		WithPartitionID(partitionID).
 		WithChannelName(vchannel).
 		WithSegmentID(segmentID).
-		WithTimeRange(ts, ts).
 		WithLevel(segmentLevel).
 		WithDataSource(metrics.BulkinsertDataSourceLabel).
 		WithBatchRows(int64(insertData.GetRowNum()))

internal/flushcommon/syncmgr/pack_writer.go

@@ -19,6 +19,7 @@ package syncmgr
 import (
 	"context"
 	"fmt"
+	"math"
 	"path"
 
 	"github.com/apache/arrow/go/v17/arrow"
@@ -330,6 +331,9 @@ func (bw *BulkPackWriter) writeDelta(ctx context.Context, pack *SyncPack) (*data
 	defer pkBuilder.Release()
 	defer tsBuilder.Release()
 
+	// Extract min/max timestamps from delete data
+	var tsFrom uint64 = math.MaxUint64
+	var tsTo uint64 = 0
 	for i := int64(0); i < pack.deltaData.RowCount; i++ {
 		switch pkField.DataType {
 		case schemapb.DataType_Int64:
@@ -339,7 +343,14 @@ func (bw *BulkPackWriter) writeDelta(ctx context.Context, pack *SyncPack) (*data
 		default:
 			return nil, fmt.Errorf("unexpected pk type %v", pkField.DataType)
 		}
-		tsBuilder.(*array.Int64Builder).Append(int64(pack.deltaData.Tss[i]))
+		ts := pack.deltaData.Tss[i]
+		tsBuilder.(*array.Int64Builder).Append(int64(ts))
+		if ts < tsFrom {
+			tsFrom = ts
+		}
+		if ts > tsTo {
+			tsTo = ts
+		}
 	}
 
 	pkArray := pkBuilder.NewArray()
@@ -362,8 +373,8 @@ func (bw *BulkPackWriter) writeDelta(ctx context.Context, pack *SyncPack) (*data
 
 	deltalog := &datapb.Binlog{
 		EntriesNum:    pack.deltaData.RowCount,
-		TimestampFrom: pack.tsFrom,
-		TimestampTo:   pack.tsTo,
+		TimestampFrom: tsFrom,
+		TimestampTo:   tsTo,
 		LogPath:       path,
 		LogSize:       pack.deltaData.Size() / 4, // Not used
 		MemorySize:    pack.deltaData.Size(),

internal/flushcommon/syncmgr/pack_writer_test.go

@@ -121,10 +121,12 @@ func TestBulkPackWriter_Write(t *testing.T) {
 				FieldID: 100,
 				Binlogs: []*datapb.Binlog{
 					{
-						EntriesNum: 10,
-						LogPath:    "files/delta_log/123/456/789/10000",
-						LogSize:    60,
-						MemorySize: 240,
+						EntriesNum:    10,
+						TimestampFrom: 100,
+						TimestampTo:   109,
+						LogPath:       "files/delta_log/123/456/789/10000",
+						LogSize:       60,
+						MemorySize:    240,
 					},
 				},
 			},

tests/integration/import/binlog_test.go

@@ -19,6 +19,7 @@ package importv2
 import (
 	"context"
 	"fmt"
+	"math"
 	"strconv"
 	"strings"
 	"time"
@@ -56,6 +57,12 @@ type SourceCollectionInfo struct {
 	SegmentIDs     []int64
 	insertedIDs    *schemapb.IDs
 	storageVersion int
+	// Timestamp ranges extracted from source segments' binlogs,
+	// there's only one L1 segment and one L0 segment after import so we can record the min and max timestamps directly.
+	l1MinTs uint64 // min timestamp from L1 segments' insert binlogs
+	l1MaxTs uint64 // max timestamp from L1 segments' insert binlogs
+	l0MinTs uint64 // min timestamp from L0 segments' delta binlogs
+	l0MaxTs uint64 // max timestamp from L0 segments' delta binlogs
 }
 
 func (s *BulkInsertSuite) PrepareSourceCollection(dim int, dmlGroup *DMLGroup) *SourceCollectionInfo {
@@ -201,6 +208,37 @@ func (s *BulkInsertSuite) PrepareSourceCollection(dim int, dmlGroup *DMLGroup) *
 		s.True(len(l0Segments) > 0)
 	}
 
+	// Extract timestamp ranges from source segments' binlogs
+	var l1MinTs uint64 = math.MaxUint64
+	var l1MaxTs uint64 = 0
+	for _, segment := range segments {
+		for _, fieldBinlog := range segment.GetBinlogs() {
+			for _, binlog := range fieldBinlog.GetBinlogs() {
+				if binlog.GetTimestampFrom() < l1MinTs {
+					l1MinTs = binlog.GetTimestampFrom()
+				}
+				if binlog.GetTimestampTo() > l1MaxTs {
+					l1MaxTs = binlog.GetTimestampTo()
+				}
+			}
+		}
+	}
+
+	var l0MinTs uint64 = math.MaxUint64
+	var l0MaxTs uint64 = 0
+	for _, segment := range l0Segments {
+		for _, fieldBinlog := range segment.GetDeltalogs() {
+			for _, binlog := range fieldBinlog.GetBinlogs() {
+				if binlog.GetTimestampFrom() < l0MinTs {
+					l0MinTs = binlog.GetTimestampFrom()
+				}
+				if binlog.GetTimestampTo() > l0MaxTs {
+					l0MaxTs = binlog.GetTimestampTo()
+				}
+			}
+		}
+	}
+
 	// search
 	expr := fmt.Sprintf("%s > 0", integration.Int64Field)
 	nq := 10
@@ -264,6 +302,10 @@ func (s *BulkInsertSuite) PrepareSourceCollection(dim int, dmlGroup *DMLGroup) *
 		}),
 		insertedIDs:    totalInsertedIDs,
 		storageVersion: int(storage.StorageV2),
+		l1MinTs:        l1MinTs,
+		l1MaxTs:        l1MaxTs,
+		l0MinTs:        l0MinTs,
+		l0MaxTs:        l0MaxTs,
 	}
 }
 
@@ -366,6 +408,13 @@ func (s *BulkInsertSuite) runBinlogTest(dmlGroup *DMLGroup) {
 	s.True(len(segment.GetStatslogs()) > 0)
 	s.NoError(CheckLogID(segment.GetStatslogs()))
 
+	// Verify L1 segment positions match actual timestamps from source collection
+	s.Equal(sourceCollectionInfo.l1MinTs, segment.GetStartPosition().GetTimestamp(),
+		"L1 segment StartPosition should match actual min timestamp from source binlogs")
+	s.Equal(sourceCollectionInfo.l1MaxTs, segment.GetDmlPosition().GetTimestamp(),
+		"L1 segment DmlPosition should match actual max timestamp from source binlogs")
+	log.Info("L1 segment position verification passed")
+
 	// l0 import
 	if totalDeleteRowNum > 0 {
 		files = make([]*internalpb.ImportFile, 0)
@@ -403,6 +452,13 @@ func (s *BulkInsertSuite) runBinlogTest(dmlGroup *DMLGroup) {
 		s.True(len(segment.GetDeltalogs()) > 0)
 		s.NoError(CheckLogID(segment.GetDeltalogs()))
 		s.True(len(segment.GetStatslogs()) == 0)
+
+		// Verify L0 segment positions match actual timestamps from source collection
+		s.Equal(sourceCollectionInfo.l0MinTs, segment.GetStartPosition().GetTimestamp(),
+			"L0 segment StartPosition should match actual min timestamp from source deltalogs")
+		s.Equal(sourceCollectionInfo.l0MaxTs, segment.GetDmlPosition().GetTimestamp(),
+			"L0 segment DmlPosition should match actual max timestamp from source deltalogs")
+		log.Info("L0 segment position verification passed")
 	}
 
 	// load