fix: generate ManifestPath for StorageV3 growing segments at creation time (#48203)

Related to #48120

When pchannel migrates between streaming nodes, growing segments lose
their ManifestPath because it was only generated locally in the write
buffer metacache, never persisted to DataCoord meta. The new node gets
an empty ManifestPath from GetSegmentInfo, causing "unexpected end of
JSON input" errors on sync.

Generate ManifestPath in openNewSegmentWithGivenSegmentID when
StorageVersion is V3, so segments are born with a valid manifest path in
DataCoord meta.

Also fix typo: UnmarshalManfestPath → UnmarshalManifestPath across
codebase.

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

internal/datacoord/meta.go

@@ -1249,7 +1249,7 @@ func UpdateManifestVersion(segmentID int64, manifestVersion int64) UpdateOperato
 				zap.Int64("segmentID", segmentID))
 			return false
 		}
-		basePath, currentVer, err := packed.UnmarshalManfestPath(segment.ManifestPath)
+		basePath, currentVer, err := packed.UnmarshalManifestPath(segment.ManifestPath)
 		if err != nil {
 			log.Ctx(context.TODO()).Warn("meta update: update manifest version failed - unmarshal error",
 				zap.Int64("segmentID", segmentID), zap.Error(err))

internal/datacoord/meta_test.go

@@ -1458,7 +1458,7 @@ func TestUpdateManifestVersion(t *testing.T) {
 		// Verify the manifest path was updated
 		seg := pack.Get(1)
 		assert.NotNil(t, seg)
-		basePath, version, err := packed.UnmarshalManfestPath(seg.ManifestPath)
+		basePath, version, err := packed.UnmarshalManifestPath(seg.ManifestPath)
 		assert.NoError(t, err)
 		assert.Equal(t, "/data/segments/1", basePath)
 		assert.Equal(t, int64(10), version)
@@ -1485,7 +1485,7 @@ func TestUpdateManifestVersion(t *testing.T) {
 
 		updated := meta.GetHealthySegment(context.TODO(), 1)
 		assert.NotNil(t, updated)
-		basePath, version, err := packed.UnmarshalManfestPath(updated.ManifestPath)
+		basePath, version, err := packed.UnmarshalManifestPath(updated.ManifestPath)
 		assert.NoError(t, err)
 		assert.Equal(t, "/data/segments/1", basePath)
 		assert.Equal(t, int64(5), version)

internal/datacoord/segment_manager.go

@@ -19,6 +19,7 @@ package datacoord
 import (
 	"context"
 	"fmt"
+	"path"
 	"sync"
 	"time"
 
@@ -29,9 +30,14 @@ import (
 
 	"github.com/milvus-io/milvus-proto/go-api/v2/commonpb"
 	"github.com/milvus-io/milvus/internal/datacoord/allocator"
+	"github.com/milvus-io/milvus/internal/storage"
+	"github.com/milvus-io/milvus/internal/storagev2/packed"
+	"github.com/milvus-io/milvus/pkg/v2/common"
 	"github.com/milvus-io/milvus/pkg/v2/log"
 	"github.com/milvus-io/milvus/pkg/v2/proto/datapb"
 	"github.com/milvus-io/milvus/pkg/v2/util/lock"
+	"github.com/milvus-io/milvus/pkg/v2/util/metautil"
+	"github.com/milvus-io/milvus/pkg/v2/util/paramtable"
 	"github.com/milvus-io/milvus/pkg/v2/util/retry"
 	"github.com/milvus-io/milvus/pkg/v2/util/tsoutil"
 	"github.com/milvus-io/milvus/pkg/v2/util/typeutil"
@@ -415,6 +421,13 @@ func (s *SegmentManager) openNewSegmentWithGivenSegmentID(ctx context.Context, r
 		}
 	}
 
+	var manifestPath string
+	if req.StorageVersion == storage.StorageV3 {
+		k := metautil.JoinIDPath(req.CollectionID, req.PartitionID, req.SegmentID)
+		basePath := path.Join(paramtable.Get().MinioCfg.RootPath.GetValue(), common.SegmentInsertLogPath, k)
+		manifestPath = packed.MarshalManifestPath(basePath, -1)
+	}
+
 	segmentInfo := &datapb.SegmentInfo{
 		ID:                   req.SegmentID,
 		CollectionID:         req.CollectionID,
@@ -427,6 +440,7 @@ func (s *SegmentManager) openNewSegmentWithGivenSegmentID(ctx context.Context, r
 		LastExpireTime:       0,
 		StorageVersion:       req.StorageVersion,
 		IsCreatedByStreaming: req.IsCreatedByStreaming,
+		ManifestPath:         manifestPath,
 	}
 	segment := NewSegmentInfo(segmentInfo)
 	if err := s.meta.AddSegment(ctx, segment); err != nil {

internal/datacoord/segment_manager_test.go

@@ -38,6 +38,7 @@ import (
 	"github.com/milvus-io/milvus/internal/metastore/kv/datacoord"
 	"github.com/milvus-io/milvus/internal/metastore/mocks"
 	"github.com/milvus-io/milvus/internal/storage"
+	"github.com/milvus-io/milvus/internal/storagev2/packed"
 	"github.com/milvus-io/milvus/pkg/v2/proto/datapb"
 	"github.com/milvus-io/milvus/pkg/v2/proto/rootcoordpb"
 	"github.com/milvus-io/milvus/pkg/v2/util/etcd"
@@ -1093,3 +1094,53 @@ func TestDropSegmentOfPartition(t *testing.T) {
 	segment = meta.GetHealthySegment(context.TODO(), segID)
 	assert.NotNil(t, segment)
 }
+
+func TestAllocNewGrowingSegment_ManifestPath(t *testing.T) {
+	ctx := context.Background()
+	paramtable.Init()
+	mockAllocator := newMockAllocator(t)
+	meta, err := newMemoryMeta(t)
+	assert.NoError(t, err)
+	segmentManager, _ := newSegmentManager(meta, mockAllocator)
+
+	schema := newTestSchema()
+	collID, err := mockAllocator.AllocID(ctx)
+	assert.NoError(t, err)
+	meta.AddCollection(&collectionInfo{ID: collID, Schema: schema})
+
+	t.Run("StorageV3 segment has manifest path", func(t *testing.T) {
+		segID, err := mockAllocator.AllocID(ctx)
+		assert.NoError(t, err)
+		segment, err := segmentManager.AllocNewGrowingSegment(ctx, AllocNewGrowingSegmentRequest{
+			CollectionID:         collID,
+			PartitionID:          100,
+			SegmentID:            segID,
+			ChannelName:          "c1",
+			StorageVersion:       storage.StorageV3,
+			IsCreatedByStreaming: true,
+		})
+		assert.NoError(t, err)
+		assert.NotEmpty(t, segment.ManifestPath)
+
+		// Verify the manifest path can be unmarshalled
+		basePath, ver, unmarshalErr := packed.UnmarshalManifestPath(segment.ManifestPath)
+		assert.NoError(t, unmarshalErr)
+		assert.NotEmpty(t, basePath)
+		assert.Equal(t, int64(-1), ver)
+	})
+
+	t.Run("StorageV2 segment has no manifest path", func(t *testing.T) {
+		segID, err := mockAllocator.AllocID(ctx)
+		assert.NoError(t, err)
+		segment, err := segmentManager.AllocNewGrowingSegment(ctx, AllocNewGrowingSegmentRequest{
+			CollectionID:         collID,
+			PartitionID:          100,
+			SegmentID:            segID,
+			ChannelName:          "c2",
+			StorageVersion:       storage.StorageV2,
+			IsCreatedByStreaming: true,
+		})
+		assert.NoError(t, err)
+		assert.Empty(t, segment.ManifestPath)
+	})
+}

internal/datanode/compactor/l0_compactor.go

@@ -263,7 +263,7 @@ func (t *LevelZeroCompactionTask) splitAndWrite(
 			if segment.GetManifest() != "" {
 				storageVersion = storage.StorageV2
 				// V3: build deltalog path under basePath/_delta/
-				basePath, _, err := packed.UnmarshalManfestPath(segment.GetManifest())
+				basePath, _, err := packed.UnmarshalManifestPath(segment.GetManifest())
 				if err != nil {
 					log.Warn("L0 compaction failed to parse manifest path", zap.Int64("segmentID", segmentID), zap.Error(err))
 					return nil, err

internal/datanode/importv2/copy_segment_utils.go

@@ -106,7 +106,7 @@ func transformManifestPath(
 	source *datapb.CopySegmentSource,
 	target *datapb.CopySegmentTarget,
 ) (string, error) {
-	basePath, version, err := packed.UnmarshalManfestPath(manifestPath)
+	basePath, version, err := packed.UnmarshalManifestPath(manifestPath)
 	if err != nil {
 		return "", fmt.Errorf("failed to unmarshal manifest path: %w", err)
 	}
@@ -215,7 +215,7 @@ func collectSegmentFiles(
 				source.GetStorageVersion(), source.GetSegmentId())
 		}
 
-		basePath, _, err := packed.UnmarshalManfestPath(manifestPath)
+		basePath, _, err := packed.UnmarshalManifestPath(manifestPath)
 		if err != nil {
 			return nil, fmt.Errorf("failed to unmarshal manifest path %q for segment %d: %w", manifestPath, source.GetSegmentId(), err)
 		}

internal/datanode/importv2/copy_segment_utils_test.go

@@ -1150,7 +1150,7 @@ func TestTransformManifestPath(t *testing.T) {
 	targetManifest, err := transformManifestPath(sourceManifest, source, target)
 	assert.NoError(t, err)
 
-	basePath, version, err := packed.UnmarshalManfestPath(targetManifest)
+	basePath, version, err := packed.UnmarshalManifestPath(targetManifest)
 	assert.NoError(t, err)
 	assert.Equal(t, int64(2), version)
 	assert.Contains(t, basePath, "111")

internal/flushcommon/syncmgr/pack_writer_v3.go

@@ -160,7 +160,7 @@ func (bw *BulkPackWriterV3) writeInsertsIntoStorage(ctx context.Context,
 		return result
 	}
 
-	basePath, version, err := packed.UnmarshalManfestPath(bw.manifestPath)
+	basePath, version, err := packed.UnmarshalManifestPath(bw.manifestPath)
 	if err != nil {
 		return nil, "", err
 	}
@@ -212,7 +212,7 @@ func (bw *BulkPackWriterV3) writeDelta(ctx context.Context, pack *SyncPack) (str
 	}
 
 	// Build deltalog path under basePath/_delta/
-	basePath, _, err := packed.UnmarshalManfestPath(bw.manifestPath)
+	basePath, _, err := packed.UnmarshalManifestPath(bw.manifestPath)
 	if err != nil {
 		return "", fmt.Errorf("failed to parse manifest path: %w", err)
 	}

internal/flushcommon/syncmgr/pack_writer_v3_test.go

@@ -171,7 +171,7 @@ func (s *PackWriterV3Suite) TestPackWriterV3_Write() {
 	gotInserts, _, _, _, writtenManifestPath, _, err := bw.Write(context.Background(), pack)
 	s.NoError(err)
 	s.Equal(gotInserts[0].Binlogs[0].GetEntriesNum(), int64(rows))
-	writtenBasePath, revision, err := packed.UnmarshalManfestPath(writtenManifestPath)
+	writtenBasePath, revision, err := packed.UnmarshalManifestPath(writtenManifestPath)
 	s.NoError(err)
 	s.Equal(basePath, writtenBasePath)
 	s.Greater(revision, int64(0))
@@ -322,7 +322,7 @@ func (s *PackWriterV3Suite) TestWriteWithDeleteData() {
 	// For V3, deltas are nil since deltalogs are stored in manifest
 	s.Nil(gotDeletes)
 	// Verify manifest was updated (version should be > -1)
-	_, revision, err := packed.UnmarshalManfestPath(writtenManifestPath)
+	_, revision, err := packed.UnmarshalManifestPath(writtenManifestPath)
 	s.NoError(err)
 	s.Greater(revision, int64(-1))
 }

internal/storagev2/packed/exttable_test.go

@@ -345,25 +345,25 @@ func TestMarshalUnmarshalManifestPath(t *testing.T) {
 	marshaled := MarshalManifestPath(original, version)
 	assert.NotEmpty(t, marshaled)
 
-	basePath, ver, err := UnmarshalManfestPath(marshaled)
+	basePath, ver, err := UnmarshalManifestPath(marshaled)
 	assert.NoError(t, err)
 	assert.Equal(t, original, basePath)
 	assert.Equal(t, version, ver)
 }
 
 func TestUnmarshalManifestPath_InvalidJSON(t *testing.T) {
-	_, _, err := UnmarshalManfestPath("not valid json")
+	_, _, err := UnmarshalManifestPath("not valid json")
 	assert.Error(t, err)
 }
 
 func TestUnmarshalManifestPath_EmptyString(t *testing.T) {
-	_, _, err := UnmarshalManfestPath("")
+	_, _, err := UnmarshalManifestPath("")
 	assert.Error(t, err)
 }
 
 func TestMarshalManifestPath_EmptyBasePath(t *testing.T) {
 	marshaled := MarshalManifestPath("", 0)
-	basePath, ver, err := UnmarshalManfestPath(marshaled)
+	basePath, ver, err := UnmarshalManifestPath(marshaled)
 	assert.NoError(t, err)
 	assert.Equal(t, "", basePath)
 	assert.Equal(t, int64(0), ver)

internal/storagev2/packed/ffi_common.go

@@ -272,7 +272,7 @@ func MarshalManifestPath(basePath string, version int64) string {
 	return string(bs)
 }
 
-func UnmarshalManfestPath(manifestPath string) (string, int64, error) {
+func UnmarshalManifestPath(manifestPath string) (string, int64, error) {
 	var manifestJSON ManifestJSON
 	err := json.Unmarshal([]byte(manifestPath), &manifestJSON)
 	if err != nil {

internal/storagev2/packed/manifest_ffi.go

@@ -181,7 +181,7 @@ func ReadFragmentsFromManifest(
 	storageConfig *indexpb.StorageConfig,
 ) ([]Fragment, error) {
 	// 1. Parse manifest path to get base path
-	basePath, _, err := UnmarshalManfestPath(manifestPath)
+	basePath, _, err := UnmarshalManifestPath(manifestPath)
 	if err != nil {
 		return nil, fmt.Errorf("failed to parse manifest path: %w", err)
 	}

internal/storagev2/packed/packed_reader_ffi.go

@@ -189,7 +189,7 @@ func (r *FFIPackedReader) Release() {
 
 func GetManifestHandle(manifestPath string, storageConfig *indexpb.StorageConfig) (loonManifestHandle *C.LoonManifest, err error) {
 	var cManifestHandle *C.LoonManifest
-	basePath, version, err := UnmarshalManfestPath(manifestPath)
+	basePath, version, err := UnmarshalManifestPath(manifestPath)
 	if err != nil {
 		return cManifestHandle, err
 	}

internal/storagev2/packed/packed_reader_ffi_test.go

@@ -373,7 +373,7 @@ func TestPackedFFIReaderMultipleBatches(t *testing.T) {
 		manifest, err = pw.Close()
 		require.NoError(t, err)
 
-		_, version, err = UnmarshalManfestPath(manifest)
+		_, version, err = UnmarshalManifestPath(manifest)
 		require.NoError(t, err)
 
 		totalWrittenRows += numRows

internal/storagev2/packed/packed_writer_ffi_test.go

@@ -105,7 +105,7 @@ func TestPackedFFIWriter(t *testing.T) {
 		require.NoError(t, err)
 		require.NotEmpty(t, manifest)
 
-		p, pv, err := UnmarshalManfestPath(manifest)
+		p, pv, err := UnmarshalManifestPath(manifest)
 		require.NoError(t, err)
 		assert.Equal(t, p, basePath)
 		assert.Equal(t, pv, version+1)

internal/storagev2/packed/transaction.go

@@ -55,7 +55,7 @@ func AddDeltaLogsToManifest(
 		return manifestPath, nil
 	}
 
-	basePath, version, err := UnmarshalManfestPath(manifestPath)
+	basePath, version, err := UnmarshalManifestPath(manifestPath)
 	if err != nil {
 		return "", fmt.Errorf("failed to parse manifest path: %w", err)
 	}
@@ -120,7 +120,7 @@ func GetDeltaLogPathsFromManifest(
 	manifestPath string,
 	storageConfig *indexpb.StorageConfig,
 ) ([]string, error) {
-	basePath, version, err := UnmarshalManfestPath(manifestPath)
+	basePath, version, err := UnmarshalManifestPath(manifestPath)
 	if err != nil {
 		return nil, fmt.Errorf("failed to parse manifest path: %w", err)
 	}