fix: force L0 import to use storage v2 (#51158)

## Summary

Fix L0 import storage version selection when `UseLoonFFI` is enabled:

- Force DataCoord to assign StorageV2 for L0 import segments.
- Force L0 import requests to disable Loon FFI.
- Defensively keep DataNode L0 sync tasks on StorageV2 even if the
request carries StorageV3.
- Add regression coverage for L0 import with `UseLoonFFI=true`.

Fixes #51148

## Known mixed-version window (documented, not fixed here)

The `AssignSegments` metadata fix only applies to jobs created on the
new version. A job created by a **pre-upgrade** DataCoord with
`common.storage.useLoonFFI=true` has already persisted its L0 segments
with `StorageVersion=3` in etcd, and segment IDs are reused (not
reallocated) when the task is recovered. If such an in-flight job
completes **after** the upgrade — either the upgraded DataCoord recovers
the task and reassembles the request as V2, or an upgraded DataNode
serves the old coordinator's V3 request (`syncDelete` now hard-codes V2)
— the deltalogs are written as V2 but the segment stays tagged
`StorageVersion=3` with an empty `ManifestPath`: the import completion
path does not reconcile `StorageVersion`, and L0 segments are excluded
from the storage-version compaction policy, so nothing self-heals the
tag.

Impact: harmless for delta load and L0 compaction (both branch on the
manifest, not the tag), but snapshot restore trusts the tag —
`collectSegmentFiles` fails on `storage_version>=3` with an empty
manifest.

Scope: requires a deployment running the experimental `useLoonFFI=true`
with an L0 import job in flight at the moment of the upgrade, later
restored via snapshot. Since the pre-upgrade V3 L0 write path always
fails (#51148), no successfully-written V3 L0 data exists — only the
metadata tag can be stale. If this window ever materializes in practice,
the backfill is a one-operator change in the completion path
(`UpdateStorageVersionOperator(segID, StorageV2)` for L0).

## Test Plan

- [x] `make static-check`
- [x] `go test -tags dynamic,test -gcflags="all=-N -l" -count=1
./internal/datacoord -run
TestImportUtil_L0ImportUsesStorageV2WhenLoonFFIEnabled`
- [x] `go test -tags dynamic,test -gcflags="all=-N -l" -count=1
./internal/datanode/importv2 -run
'TestL0Import/TestL0ImportForcesStorageV2ForSyncTask'`

Note: after rebasing to latest master, the focused local tests could not
be rerun with the reused local C++ output because the local
`libmilvus_core` is older than current headers and misses
`_SegcoreSetPrefetchThreadPoolNum`. The same focused tests passed before
the rebase; CI should build with matching artifacts.

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

internal/datacoord/import_util.go

@@ -145,6 +145,20 @@ func GetSegmentMaxSize(job ImportJob, meta *meta) int {
 	return int(getExpectedSegmentSize(meta, job.GetCollectionID(), job.GetSchema()))
 }
 
+func importStorageVersion(isL0Import bool) int64 {
+	if isL0Import {
+		return storage.StorageV2
+	}
+	if paramtable.Get().CommonCfg.UseLoonFFI.GetAsBool() {
+		return storage.StorageV3
+	}
+	return storage.StorageV2
+}
+
+func importUseLoonFFI(isL0Import bool) bool {
+	return !isL0Import && paramtable.Get().CommonCfg.UseLoonFFI.GetAsBool()
+}
+
 func AssignSegments(job ImportJob, task ImportTask, alloc allocator.Allocator, meta *meta, segmentMaxSize int64) ([]int64, error) {
 	pkField, err := typeutil.GetPrimaryFieldSchema(job.GetSchema())
 	if err != nil {
@@ -170,10 +184,7 @@ func AssignSegments(job ImportJob, task ImportTask, alloc allocator.Allocator, m
 		segmentLevel = datapb.SegmentLevel_L0
 	}
 
-	storageVersion := storage.StorageV2
-	if paramtable.Get().CommonCfg.UseLoonFFI.GetAsBool() {
-		storageVersion = storage.StorageV3
-	}
+	storageVersion := importStorageVersion(isL0Import)
 
 	// alloc new segments
 	segments := make([]int64, 0)
@@ -350,10 +361,9 @@ func AssembleImportRequest(task ImportTask, job ImportJob, meta *meta, alloc all
 		return fileStat.GetImportFile()
 	})
 
-	storageVersion := storage.StorageV2
-	if paramtable.Get().CommonCfg.UseLoonFFI.GetAsBool() {
-		storageVersion = storage.StorageV3
-	}
+	isL0Import := importutilv2.IsL0Import(job.GetOptions())
+	storageVersion := importStorageVersion(isL0Import)
+	useLoonFFI := importUseLoonFFI(isL0Import)
 
 	req := &datapb.ImportRequest{
 		ClusterID:       Params.CommonCfg.ClusterPrefix.GetValue(),
@@ -372,7 +382,7 @@ func AssembleImportRequest(task ImportTask, job ImportJob, meta *meta, alloc all
 		TaskSlot:        task.GetTaskSlot(),
 		StorageVersion:  storageVersion,
 		PluginContext:   GetReadPluginContext(job.GetOptions()),
-		UseLoonFfi:      Params.CommonCfg.UseLoonFFI.GetAsBool(),
+		UseLoonFfi:      useLoonFFI,
 	}
 	WrapPluginContext(task.GetCollectionID(), job.GetSchema().GetProperties(), req)
 	return req, nil

internal/datacoord/import_util_test.go

@@ -380,6 +380,90 @@ func TestImportUtil_AssembleRequestWithDataTt(t *testing.T) {
 	assert.Equal(t, job.GetVchannels(), importReq.GetVchannels())
 }
 
+func TestImportUtil_L0ImportUsesStorageV2WhenLoonFFIEnabled(t *testing.T) {
+	paramtable.Get().Save(paramtable.Get().CommonCfg.UseLoonFFI.Key, "true")
+	defer paramtable.Get().Reset(paramtable.Get().CommonCfg.UseLoonFFI.Key)
+
+	job := &importJob{
+		ImportJob: &datapb.ImportJob{
+			JobID:        1,
+			CollectionID: 2,
+			PartitionIDs: []int64{3},
+			Vchannels:    []string{"c0"},
+			Options: []*commonpb.KeyValuePair{
+				{Key: importutilv2.L0Import, Value: "true"},
+			},
+			Schema: &schemapb.CollectionSchema{
+				Fields: []*schemapb.FieldSchema{
+					{
+						FieldID:      100,
+						Name:         "pk",
+						DataType:     schemapb.DataType_Int64,
+						IsPrimaryKey: true,
+					},
+				},
+			},
+		},
+	}
+	taskProto := &datapb.ImportTaskV2{
+		JobID:        job.GetJobID(),
+		TaskID:       4,
+		CollectionID: job.GetCollectionID(),
+		FileStats: []*datapb.ImportFileStats{
+			{
+				ImportFile: &internalpb.ImportFile{Id: 0, Paths: []string{"l0-prefix"}},
+				HashedStats: map[string]*datapb.PartitionImportStats{
+					"c0": {PartitionDataSize: map[int64]int64{3: 1}},
+				},
+			},
+		},
+	}
+	importMeta := NewMockImportMeta(t)
+	importMeta.EXPECT().GetJob(mock.Anything, mock.Anything).Return(job)
+	task := &importTask{
+		importMeta: importMeta,
+	}
+	task.task.Store(taskProto)
+
+	alloc := allocator.NewMockAllocator(t)
+	alloc.EXPECT().AllocID(mock.Anything).Return(int64(10), nil)
+	alloc.EXPECT().AllocTimestamp(mock.Anything).Return(uint64(100), nil)
+	alloc.EXPECT().AllocN(mock.Anything).RunAndReturn(func(n int64) (int64, int64, error) {
+		return 1000, 1000 + n, nil
+	})
+
+	catalog := mocks.NewDataCoordCatalog(t)
+	catalog.EXPECT().ListChannelCheckpoint(mock.Anything).Return(nil, nil)
+	catalog.EXPECT().ListIndexes(mock.Anything).Return(nil, nil)
+	catalog.EXPECT().ListSegmentIndexes(mock.Anything, mock.Anything).Return(nil, nil).Maybe()
+	catalog.EXPECT().AddSegment(mock.Anything, mock.Anything).Return(nil)
+	catalog.EXPECT().ListAnalyzeTasks(mock.Anything).Return(nil, nil)
+	catalog.EXPECT().ListCompactionTask(mock.Anything).Return(nil, nil)
+	catalog.EXPECT().ListPartitionStatsInfos(mock.Anything).Return(nil, nil)
+	catalog.EXPECT().ListStatsTasks(mock.Anything).Return(nil, nil)
+	catalog.EXPECT().ListSnapshots(mock.Anything).Return(nil, nil)
+	catalog.EXPECT().ListExternalCollectionRefreshJobs(mock.Anything).Return(nil, nil)
+	catalog.EXPECT().ListExternalCollectionRefreshTasks(mock.Anything).Return(nil, nil)
+
+	broker := broker.NewMockBroker(t)
+	broker.EXPECT().ShowCollectionIDs(mock.Anything).Return(nil, nil)
+	meta, err := newMeta(context.TODO(), catalog, nil, broker)
+	assert.NoError(t, err)
+
+	segments, err := AssignSegments(job, task, alloc, meta, 1024)
+	assert.NoError(t, err)
+	assert.Equal(t, []int64{10}, segments)
+	segment := meta.GetSegment(context.Background(), 10)
+	assert.NotNil(t, segment)
+	assert.Equal(t, datapb.SegmentLevel_L0, segment.GetLevel())
+	assert.EqualValues(t, storage.StorageV2, segment.GetStorageVersion())
+
+	importReq, err := AssembleImportRequest(task, job, meta, alloc)
+	assert.NoError(t, err)
+	assert.EqualValues(t, storage.StorageV2, importReq.GetStorageVersion())
+	assert.False(t, importReq.GetUseLoonFfi())
+}
+
 func TestImportUtil_RegroupImportFiles(t *testing.T) {
 	fileNum := 4096
 	dataSize := paramtable.Get().DataCoordCfg.SegmentMaxSize.GetAsInt64() * 1024 * 1024

internal/datanode/importv2/task_l0_import.go

@@ -252,7 +252,7 @@ func (t *L0ImportTask) syncDelete(delData []*storage.DeleteData) ([]*conc.Future
 		}
 		syncTask, err := NewSyncTask(t.ctx, t.allocator, t.metaCaches, t.req.GetTs(),
 			segmentID, partitionID, t.GetCollectionID(), channel, nil, data,
-			nil, t.req.GetStorageVersion(), false, t.req.GetStorageConfig())
+			nil, storage.StorageV2, false, t.req.GetStorageConfig())
 		if err != nil {
 			return nil, nil, err
 		}

internal/datanode/importv2/task_l0_import_test.go

@@ -94,14 +94,14 @@ func (s *L0ImportSuite) SetupTest() {
 	s.NoError(err)
 
 	cm := mocks.NewChunkManager(s.T())
-	cm.EXPECT().MultiRead(mock.Anything, mock.Anything).Return([][]byte{blob.Value}, nil)
+	cm.EXPECT().MultiRead(mock.Anything, mock.Anything).Return([][]byte{blob.Value}, nil).Maybe()
 	cm.EXPECT().WalkWithPrefix(mock.Anything, mock.Anything, mock.Anything, mock.Anything).RunAndReturn(
 		func(ctx context.Context, s string, b bool, walkFunc storage.ChunkObjectWalkFunc) error {
 			for _, file := range []string{"a/b/c/"} {
 				walkFunc(&storage.ChunkObjectInfo{FilePath: file})
 			}
 			return nil
-		})
+		}).Maybe()
 	s.cm = cm
 }
 
@@ -194,6 +194,50 @@ func (s *L0ImportSuite) TestL0Import() {
 	// s.Equal(s.deleteData.Size(), deltaLog.GetMemorySize())
 }
 
+func (s *L0ImportSuite) TestL0ImportForcesStorageV2ForSyncTask() {
+	var task *L0ImportTask
+	s.syncMgr.EXPECT().SyncDataWithChunkManager(mock.Anything, mock.Anything, mock.Anything).
+		RunAndReturn(func(ctx context.Context, syncTask syncmgr.Task, cm storage.ChunkManager, callbacks ...func(error) error) (*conc.Future[struct{}], error) {
+			segment, ok := task.metaCaches[s.channel].GetSegmentByID(s.segmentID)
+			s.True(ok)
+			s.EqualValues(storage.StorageV2, segment.GetStorageVersion())
+			s.Empty(segment.ManifestPath())
+
+			future := conc.Go(func() (struct{}, error) {
+				return struct{}{}, nil
+			})
+			return future, nil
+		})
+
+	req := &datapb.ImportRequest{
+		JobID:          1,
+		TaskID:         2,
+		CollectionID:   s.collectionID,
+		PartitionIDs:   []int64{s.partitionID},
+		Vchannels:      []string{s.channel},
+		Schema:         s.schema,
+		StorageVersion: storage.StorageV3,
+		UseLoonFfi:     true,
+		RequestSegments: []*datapb.ImportRequestSegment{
+			{
+				SegmentID:   s.segmentID,
+				PartitionID: s.partitionID,
+				Vchannel:    s.channel,
+			},
+		},
+		IDRange: &datapb.IDRange{
+			Begin: 0,
+			End:   int64(s.delCnt),
+		},
+	}
+	task = NewL0ImportTask(req, s.manager, s.syncMgr, s.cm).(*L0ImportTask)
+
+	futures, syncTasks, err := task.syncDelete([]*storage.DeleteData{s.deleteData})
+	s.NoError(err)
+	s.Len(futures, 1)
+	s.Len(syncTasks, 1)
+}
+
 func TestL0Import(t *testing.T) {
 	suite.Run(t, new(L0ImportSuite))
 }