feat: add function field backfill - Part 4: streaming part (#48809)(#48808) (#48865)

## Summary - Add schema version check on insert path in streaming node shard interceptor, rejecting inserts with mismatched schema version - Add `AlterCollection`/`AppendNewCollectionSchema`/`CheckIfCollectionSchemaVersionMatch` methods to shard manager for schema lifecycle management - Propagate schema version through segment allocation chain: `shard_manager_segment` → `partition_manager` → `segment_alloc_worker` → `CreateSegment` message header - Add `SchemaVersionMismatch` streaming error type (classified as unrecoverable to force proxy metadata refresh) - Track collection schema in `CollectionInfo` with `CollectionSchemaOfVChannel` for version gating related: https://github.com/milvus-io/milvus/issues/48808 design doc: https://github.com/milvus-io/milvus-design-docs/blob/main/design_docs/20260129-add-function-field-design.md ## Test plan - [x] partition_manager_test.go updated for new `schemaVersion` parameter - [x] wal_test.go updated with CollectionSchema in CreateCollection messages - [ ] CI: code-check, ut-go, integration-test 🤖 Generated with [Claude Code](https://claude.ai/code) Signed-off-by: MrPresent-Han <chun.han@gmail.com> Co-authored-by: MrPresent-Han <chun.han@gmail.com> Co-authored-by: Claude Opus 4.6 <noreply@anthropic.com>
2026-07-21 10:15:43 +00:00 · 2026-04-23 23:29:46 +08:00
parent 5030ae2520
commit fa0343b580
30 changed files with 1250 additions and 168 deletions
@@ -285,6 +285,7 @@ func getServiceWithChannel(initCtx context.Context, params *util.PipelineParams,
 	dmStreamNode := newDmInputNode(config, input)
 	nodeList = append(nodeList, dmStreamNode)

+	// 1.ddNode
 	ddNode := newDDNode(
 		params.Ctx,
 		collectionID,
@@ -296,20 +297,21 @@ func getServiceWithChannel(initCtx context.Context, params *util.PipelineParams,
 	)
 	nodeList = append(nodeList, ddNode)

-	if len(info.GetSchema().GetFunctions()) > 0 {
-		emNode, err := newEmbeddingNode(channelName, config.metacache)
-		if err != nil {
-			return nil, err
-		}
-		nodeList = append(nodeList, emNode)
+	// 2.embeddingNode(maybe no function)
+	emNode, err := newEmbeddingNode(channelName, config.metacache)
+	if err != nil {
+		return nil, err
 	}
+	nodeList = append(nodeList, emNode)

+	// 3.writeNode
 	writeNode, err := newWriteNode(params.Ctx, params.WriteBufferManager, ds.timetickSender, config)
 	if err != nil {
 		return nil, err
 	}
 	nodeList = append(nodeList, writeNode)

+	// 4.ttNode
 	ttNode := newTTNode(config, params.WriteBufferManager, params.CheckpointUpdater)
 	nodeList = append(nodeList, ttNode)

@@ -485,7 +485,7 @@ func (s *DataSyncServiceSuite) TestStartStop() {
 	}
 	timeTickMsgPack.Msgs = append(timeTickMsgPack.Msgs, timeTickMsg)

-	s.wbManager.EXPECT().BufferData(insertChannelName, mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(nil)
+	s.wbManager.EXPECT().BufferData(insertChannelName, mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(nil)
 	s.wbManager.EXPECT().GetCheckpoint(insertChannelName).Return(&msgpb.MsgPosition{Timestamp: msgTs, ChannelName: insertChannelName, MsgID: []byte{0}}, true, nil)
 	s.wbManager.EXPECT().NotifyCheckpointUpdated(insertChannelName, msgTs).Return().Maybe()

@@ -18,6 +18,7 @@ package pipeline

 import (
 	"fmt"
+	"strconv"

 	"github.com/cockroachdb/errors"
 	"github.com/samber/lo"
@@ -29,6 +30,7 @@ import (
 	"github.com/milvus-io/milvus/internal/storage"
 	"github.com/milvus-io/milvus/internal/util/function"
 	"github.com/milvus-io/milvus/pkg/v2/log"
+	"github.com/milvus-io/milvus/pkg/v2/util/merr"
 	"github.com/milvus-io/milvus/pkg/v2/util/paramtable"
 )

@@ -43,7 +45,9 @@ type embeddingNode struct {
 	channelName string

 	// embeddingType EmbeddingType
+	// curSchema and functionRunners are updated only from Operate (single flow-graph worker).
 	functionRunners map[int64]function.FunctionRunner
+	curSchema       *schemapb.CollectionSchema
 }

 func newEmbeddingNode(channelName string, metaCache metacache.MetaCache) (*embeddingNode, error) {
@@ -51,15 +55,15 @@ func newEmbeddingNode(channelName string, metaCache metacache.MetaCache) (*embed
 	baseNode.SetMaxQueueLength(paramtable.Get().DataNodeCfg.FlowGraphMaxQueueLength.GetAsInt32())
 	baseNode.SetMaxParallelism(paramtable.Get().DataNodeCfg.FlowGraphMaxParallelism.GetAsInt32())

+	schema := metaCache.GetSchema(0)
 	node := &embeddingNode{
 		BaseNode:        baseNode,
 		channelName:     channelName,
 		metaCache:       metaCache,
 		functionRunners: make(map[int64]function.FunctionRunner),
+		curSchema:       schema,
 	}

-	schema := metaCache.GetSchema(0)
-
 	for _, field := range schema.GetFields() {
 		if field.GetIsPrimaryKey() {
 			node.pkField = field
@@ -67,16 +71,11 @@ func newEmbeddingNode(channelName string, metaCache metacache.MetaCache) (*embed
 		}
 	}

-	for _, tf := range schema.GetFunctions() {
-		functionRunner, err := function.NewFunctionRunner(schema, tf)
-		if err != nil {
-			return nil, err
-		}
-		if functionRunner == nil {
-			continue
-		}
-		node.functionRunners[tf.GetId()] = functionRunner
+	runners, err := buildFunctionRunners(schema)
+	if err != nil {
+		return nil, err
 	}
+	node.functionRunners = runners
 	return node, nil
 }

@@ -179,7 +178,8 @@ func (eNode *embeddingNode) embedding(datas []*storage.InsertData) (map[int64]*s
 					return nil, err
 				}
 			default:
-				return nil, fmt.Errorf("unknown function type %s", functionSchema.Type)
+				return nil, merr.WrapErrServiceInternal("unknown function type in embedding node",
+					"type="+strconv.FormatInt(int64(functionSchema.GetType()), 10))
 			}
 		}
 	}
@@ -199,24 +199,88 @@ func (eNode *embeddingNode) Embedding(datas []*writebuffer.InsertData) error {
 	return nil
 }

+func buildFunctionRunners(schema *schemapb.CollectionSchema) (map[int64]function.FunctionRunner, error) {
+	runners := make(map[int64]function.FunctionRunner)
+	for _, tf := range schema.GetFunctions() {
+		functionRunner, err := function.NewFunctionRunner(schema, tf)
+		if err != nil {
+			closeFunctionRunners(runners)
+			return nil, err
+		}
+		if functionRunner == nil {
+			continue
+		}
+		runners[tf.GetId()] = functionRunner
+	}
+	return runners, nil
+}
+
+func closeFunctionRunners(runners map[int64]function.FunctionRunner) {
+	for _, r := range runners {
+		if r != nil {
+			r.Close()
+		}
+	}
+}
+
+// syncFunctionRunners rebuilds runners when the collection schema pointer changes.
+// Replacement is atomic: on error the previous schema and runners are preserved.
+func (eNode *embeddingNode) syncFunctionRunners(schema *schemapb.CollectionSchema) (bool, error) {
+	if schema == eNode.curSchema {
+		return len(eNode.functionRunners) > 0, nil
+	}
+
+	newRunners, err := buildFunctionRunners(schema)
+	if err != nil {
+		return false, err
+	}
+
+	closeFunctionRunners(eNode.functionRunners)
+	eNode.functionRunners = newRunners
+	eNode.curSchema = schema
+	return len(newRunners) > 0, nil
+}
+
 func (eNode *embeddingNode) Operate(in []Msg) []Msg {
 	fgMsg := in[0].(*FlowGraphMsg)

 	if fgMsg.IsCloseMsg() {
 		return []Msg{fgMsg}
 	}
+	currentSchema := eNode.metaCache.GetSchema(fgMsg.TimeTick())
+	hasFunctions, err := eNode.syncFunctionRunners(currentSchema)
+	if err != nil {
+		log.Error("embedding node: sync function runners failed",
+			zap.Int64("collectionID", eNode.metaCache.Collection()),
+			zap.String("channel", eNode.channelName),
+			zap.Uint64("timeTick", fgMsg.TimeTick()),
+			zap.Error(err))
+		panic(err)
+	}
+	if !hasFunctions {
+		return []Msg{fgMsg}
+	}
+	if len(fgMsg.InsertMessages) == 0 {
+		fgMsg.InsertData = make([]*writebuffer.InsertData, 0)
+		return []Msg{fgMsg}
+	}

-	insertData := make([]*writebuffer.InsertData, 0)
-	if len(fgMsg.InsertMessages) > 0 {
-		var err error
-		if insertData, err = writebuffer.PrepareInsert(eNode.metaCache.GetSchema(fgMsg.TimeTick()), eNode.pkField, fgMsg.InsertMessages); err != nil {
-			log.Error("failed to prepare insert data", zap.Error(err))
-			panic(err)
-		}
+	insertData, err := writebuffer.PrepareInsert(currentSchema, eNode.pkField, fgMsg.InsertMessages)
+	if err != nil {
+		log.Error("embedding node: prepare insert data failed",
+			zap.Int64("collectionID", eNode.metaCache.Collection()),
+			zap.String("channel", eNode.channelName),
+			zap.Uint64("timeTick", fgMsg.TimeTick()),
+			zap.Error(err))
+		panic(err)
 	}

 	if err := eNode.Embedding(insertData); err != nil {
-		log.Warn("failed to embedding insert data", zap.Error(err))
+		log.Error("embedding node: embedding insert data failed",
+			zap.Int64("collectionID", eNode.metaCache.Collection()),
+			zap.String("channel", eNode.channelName),
+			zap.Uint64("timeTick", fgMsg.TimeTick()),
+			zap.Error(err))
 		panic(err)
 	}

@@ -31,6 +31,211 @@ import (
 	"github.com/milvus-io/milvus/pkg/v2/mq/msgstream"
 )

+// makeBM25Schema returns a schema with one BM25 function field, reused across tests.
+func makeBM25Schema() *schemapb.CollectionSchema {
+	return &schemapb.CollectionSchema{
+		Fields: []*schemapb.FieldSchema{
+			{
+				FieldID:  common.TimeStampField,
+				Name:     common.TimeStampFieldName,
+				DataType: schemapb.DataType_Int64,
+			},
+			{
+				Name:         "pk",
+				FieldID:      100,
+				IsPrimaryKey: true,
+				DataType:     schemapb.DataType_Int64,
+			},
+			{
+				Name:     "text",
+				FieldID:  101,
+				DataType: schemapb.DataType_VarChar,
+				TypeParams: []*commonpb.KeyValuePair{
+					{Key: "enable_analyzer", Value: "true"},
+				},
+			},
+			{
+				Name:             "sparse",
+				FieldID:          102,
+				DataType:         schemapb.DataType_SparseFloatVector,
+				IsFunctionOutput: true,
+			},
+		},
+		Functions: []*schemapb.FunctionSchema{{
+			Name:           "BM25",
+			Type:           schemapb.FunctionType_BM25,
+			InputFieldIds:  []int64{101},
+			OutputFieldIds: []int64{102},
+		}},
+	}
+}
+
+// makePlainSchema returns a schema with no function fields.
+func makePlainSchema() *schemapb.CollectionSchema {
+	return &schemapb.CollectionSchema{
+		Fields: []*schemapb.FieldSchema{
+			{FieldID: common.TimeStampField, Name: common.TimeStampFieldName, DataType: schemapb.DataType_Int64},
+			{FieldID: 100, Name: "pk", IsPrimaryKey: true, DataType: schemapb.DataType_Int64},
+		},
+	}
+}
+
+func TestBuildFunctionRunners(t *testing.T) {
+	t.Run("schema with no functions returns empty map", func(t *testing.T) {
+		runners, err := buildFunctionRunners(makePlainSchema())
+		assert.NoError(t, err)
+		assert.Empty(t, runners)
+	})
+
+	t.Run("schema with BM25 function returns one runner", func(t *testing.T) {
+		runners, err := buildFunctionRunners(makeBM25Schema())
+		assert.NoError(t, err)
+		assert.Len(t, runners, 1)
+		closeFunctionRunners(runners)
+	})
+}
+
+func TestCloseFunctionRunners(t *testing.T) {
+	t.Run("nil map does not panic", func(t *testing.T) {
+		assert.NotPanics(t, func() { closeFunctionRunners(nil) })
+	})
+
+	t.Run("empty map does not panic", func(t *testing.T) {
+		emptyRunners, err := buildFunctionRunners(makePlainSchema())
+		assert.NoError(t, err)
+		assert.NotPanics(t, func() { closeFunctionRunners(emptyRunners) })
+	})
+
+	t.Run("map with valid runner closes cleanly", func(t *testing.T) {
+		runners, err := buildFunctionRunners(makeBM25Schema())
+		assert.NoError(t, err)
+		assert.NotPanics(t, func() { closeFunctionRunners(runners) })
+	})
+}
+
+func TestSyncFunctionRunners(t *testing.T) {
+	schema1 := makeBM25Schema()
+
+	metaCache := metacache.NewMockMetaCache(t)
+	metaCache.EXPECT().GetSchema(mock.Anything).Return(schema1)
+	metaCache.EXPECT().Collection().Return(int64(1)).Maybe()
+
+	node, err := newEmbeddingNode("test-ch", metaCache)
+	assert.NoError(t, err)
+	defer node.Free()
+
+	t.Run("same schema pointer is a no-op", func(t *testing.T) {
+		prevRunners := node.functionRunners
+		has, err := node.syncFunctionRunners(schema1)
+		assert.NoError(t, err)
+		assert.True(t, has)
+		// runners must not be replaced when schema pointer is unchanged
+		assert.Equal(t, prevRunners, node.functionRunners)
+	})
+
+	t.Run("new schema with no functions clears runners", func(t *testing.T) {
+		schema2 := makePlainSchema()
+		has, err := node.syncFunctionRunners(schema2)
+		assert.NoError(t, err)
+		assert.False(t, has)
+		assert.Equal(t, schema2, node.curSchema)
+		assert.Empty(t, node.functionRunners)
+	})
+
+	t.Run("new schema with BM25 function rebuilds runners", func(t *testing.T) {
+		schema3 := makeBM25Schema()
+		has, err := node.syncFunctionRunners(schema3)
+		assert.NoError(t, err)
+		assert.True(t, has)
+		assert.Equal(t, schema3, node.curSchema)
+		assert.Len(t, node.functionRunners, 1)
+	})
+}
+
+func TestEmbeddingNode_OperateWithSchemaChange(t *testing.T) {
+	plainSchema := makePlainSchema()
+	bm25Schema := makeBM25Schema()
+
+	t.Run("no functions schema returns msg unchanged", func(t *testing.T) {
+		metaCache := metacache.NewMockMetaCache(t)
+		metaCache.EXPECT().GetSchema(mock.Anything).Return(plainSchema)
+		metaCache.EXPECT().Collection().Return(int64(1)).Maybe()
+
+		node, err := newEmbeddingNode("ch", metaCache)
+		assert.NoError(t, err)
+		defer node.Free()
+
+		fgMsg := &FlowGraphMsg{
+			BaseMsg: flowgraph.NewBaseMsg(false),
+			InsertMessages: []*msgstream.InsertMsg{{
+				BaseMsg:       msgstream.BaseMsg{},
+				InsertRequest: &msgpb.InsertRequest{},
+			}},
+		}
+		out := node.Operate([]Msg{fgMsg})
+		assert.Len(t, out, 1)
+		// InsertData must not be set — embedding was skipped
+		assert.Nil(t, out[0].(*FlowGraphMsg).InsertData)
+	})
+
+	t.Run("functions present but no insert messages returns empty InsertData", func(t *testing.T) {
+		metaCache := metacache.NewMockMetaCache(t)
+		metaCache.EXPECT().GetSchema(mock.Anything).Return(bm25Schema)
+		metaCache.EXPECT().Collection().Return(int64(1)).Maybe()
+
+		node, err := newEmbeddingNode("ch", metaCache)
+		assert.NoError(t, err)
+		defer node.Free()
+
+		fgMsg := &FlowGraphMsg{
+			BaseMsg:        flowgraph.NewBaseMsg(false),
+			InsertMessages: nil,
+		}
+		out := node.Operate([]Msg{fgMsg})
+		assert.Len(t, out, 1)
+		msg := out[0].(*FlowGraphMsg)
+		assert.NotNil(t, msg.InsertData)
+		assert.Empty(t, msg.InsertData)
+	})
+
+	t.Run("schema changes from no-function to BM25 mid-stream", func(t *testing.T) {
+		// GetSchema is called once during newEmbeddingNode construction, then once per Operate.
+		// calls 1+2 → plainSchema; call 3 → bm25Schema.
+		callCount := 0
+		metaCache := metacache.NewMockMetaCache(t)
+		metaCache.EXPECT().GetSchema(mock.Anything).RunAndReturn(func(_ uint64) *schemapb.CollectionSchema {
+			callCount++
+			if callCount <= 2 {
+				return plainSchema
+			}
+			return bm25Schema
+		})
+		metaCache.EXPECT().Collection().Return(int64(1)).Maybe()
+
+		node, err := newEmbeddingNode("ch", metaCache)
+		assert.NoError(t, err)
+		defer node.Free()
+
+		// First call: plain schema — no functions, embedding skipped
+		out1 := node.Operate([]Msg{&FlowGraphMsg{
+			BaseMsg:        flowgraph.NewBaseMsg(false),
+			InsertMessages: []*msgstream.InsertMsg{{}},
+		}})
+		assert.Nil(t, out1[0].(*FlowGraphMsg).InsertData)
+
+		// Second call: BM25 schema — functions present, no messages → empty InsertData
+		out2 := node.Operate([]Msg{&FlowGraphMsg{
+			BaseMsg:        flowgraph.NewBaseMsg(false),
+			InsertMessages: nil,
+		}})
+		msg2 := out2[0].(*FlowGraphMsg)
+		assert.NotNil(t, msg2.InsertData)
+		assert.Empty(t, msg2.InsertData)
+		// node must now track the BM25 schema
+		assert.Equal(t, bm25Schema, node.curSchema)
+	})
+}
+
 func TestEmbeddingNode_Operator(t *testing.T) {
 	// Define test cases for different function types
 	testCases := []struct {
@@ -140,6 +345,7 @@ func TestEmbeddingNode_Operator(t *testing.T) {

 			metaCache := metacache.NewMockMetaCache(t)
 			metaCache.EXPECT().GetSchema(mock.Anything).Return(collSchema)
+			metaCache.EXPECT().Collection().Return(int64(0)).Maybe()

 			t.Run("normal case", func(t *testing.T) {
 				node, err := newEmbeddingNode("test-channel", metaCache)
@@ -80,12 +80,14 @@ func (wNode *writeNode) Operate(in []Msg) []Msg {
 	}()

 	start, end := fgMsg.StartPositions[0], fgMsg.EndPositions[0]
+	currentSchema := wNode.metacache.GetSchema(fgMsg.TimeTick())
+	schemaVersion := currentSchema.GetVersion()

 	if fgMsg.InsertData == nil {
 		insertData := make([]*writebuffer.InsertData, 0)
 		if len(fgMsg.InsertMessages) > 0 {
 			var err error
-			if insertData, err = writebuffer.PrepareInsert(wNode.metacache.GetSchema(fgMsg.TimeTick()), wNode.pkField, fgMsg.InsertMessages); err != nil {
+			if insertData, err = writebuffer.PrepareInsert(currentSchema, wNode.pkField, fgMsg.InsertMessages); err != nil {
 				log.Error("failed to prepare data", zap.Error(err))
 				panic(err)
 			}
@@ -93,7 +95,7 @@ func (wNode *writeNode) Operate(in []Msg) []Msg {
 		fgMsg.InsertData = insertData
 	}

-	err := wNode.wbManager.BufferData(wNode.channelName, fgMsg.InsertData, fgMsg.DeleteMessages, start, end)
+	err := wNode.wbManager.BufferData(wNode.channelName, fgMsg.InsertData, fgMsg.DeleteMessages, start, end, schemaVersion)
 	if err != nil {
 		log.Error("failed to buffer data", zap.Error(err))
 		panic(err)
@@ -59,13 +59,13 @@ func (wb *l0WriteBuffer) dispatchDeleteMsgsWithoutFilter(deleteMsgs []*msgstream
 	}
 }

-func (wb *l0WriteBuffer) BufferData(insertData []*InsertData, deleteMsgs []*msgstream.DeleteMsg, startPos, endPos *msgpb.MsgPosition) error {
+func (wb *l0WriteBuffer) BufferData(insertData []*InsertData, deleteMsgs []*msgstream.DeleteMsg, startPos, endPos *msgpb.MsgPosition, schemaVersion int32) error {
 	wb.mut.Lock()
 	defer wb.mut.Unlock()

 	// buffer insert data and add segment if not exists
 	for _, inData := range insertData {
-		err := wb.bufferInsert(inData, startPos, endPos)
+		err := wb.bufferInsert(inData, startPos, endPos, schemaVersion)
 		if err != nil {
 			return err
 		}
@@ -91,8 +91,8 @@ func (wb *l0WriteBuffer) BufferData(insertData []*InsertData, deleteMsgs []*msgs
 }

 // bufferInsert function InsertMsg into bufferred InsertData and returns primary key field data for future usage.
-func (wb *l0WriteBuffer) bufferInsert(inData *InsertData, startPos, endPos *msgpb.MsgPosition) error {
-	wb.CreateNewGrowingSegment(inData.partitionID, inData.segmentID, startPos)
+func (wb *l0WriteBuffer) bufferInsert(inData *InsertData, startPos, endPos *msgpb.MsgPosition, schemaVersion int32) error {
+	wb.CreateNewGrowingSegment(inData.partitionID, inData.segmentID, startPos, schemaVersion)
 	segBuf := wb.getOrCreateBuffer(inData.segmentID, startPos.GetTimestamp())

 	totalMemSize := segBuf.insertBuffer.Buffer(inData, startPos, endPos)
@@ -190,7 +190,7 @@ func (s *L0WriteBufferSuite) TestBufferData() {
 		metrics.DataNodeFlowGraphBufferDataSize.Reset()
 		insertData, err := PrepareInsert(s.collSchema, s.pkSchema, []*msgstream.InsertMsg{msg})
 		s.NoError(err)
-		err = wb.BufferData(insertData, []*msgstream.DeleteMsg{delMsg}, &msgpb.MsgPosition{Timestamp: 100}, &msgpb.MsgPosition{Timestamp: 200})
+		err = wb.BufferData(insertData, []*msgstream.DeleteMsg{delMsg}, &msgpb.MsgPosition{Timestamp: 100}, &msgpb.MsgPosition{Timestamp: 200}, 100)
 		s.NoError(err)

 		value, err := metrics.DataNodeFlowGraphBufferDataSize.GetMetricWithLabelValues(paramtable.GetStringNodeID(), fmt.Sprint(s.metacache.Collection()))
@@ -198,7 +198,7 @@ func (s *L0WriteBufferSuite) TestBufferData() {
 		s.MetricsEqual(value, 5616)

 		delMsg = s.composeDeleteMsg(lo.Map(pks, func(id int64, _ int) storage.PrimaryKey { return storage.NewInt64PrimaryKey(id) }))
-		err = wb.BufferData([]*InsertData{}, []*msgstream.DeleteMsg{delMsg}, &msgpb.MsgPosition{Timestamp: 100}, &msgpb.MsgPosition{Timestamp: 200})
+		err = wb.BufferData([]*InsertData{}, []*msgstream.DeleteMsg{delMsg}, &msgpb.MsgPosition{Timestamp: 100}, &msgpb.MsgPosition{Timestamp: 200}, 100)
 		s.NoError(err)
 		s.MetricsEqual(value, 5856)
 	})
@@ -27,7 +27,7 @@ type BufferManager interface {
 	// Register adds a WriteBuffer with provided schema & options.
 	Register(channel string, metacache metacache.MetaCache, opts ...WriteBufferOption) error
 	// CreateNewGrowingSegment notifies writeBuffer to create a new growing segment.
-	CreateNewGrowingSegment(ctx context.Context, channel string, partition int64, segmentID int64) error
+	CreateNewGrowingSegment(ctx context.Context, channel string, partition int64, segmentID int64, schemaVersion int32) error
 	// SealSegments notifies writeBuffer corresponding to provided channel to seal segments.
 	// which will cause segment start flush procedure.
 	SealSegments(ctx context.Context, channel string, segmentIDs []int64) error
@@ -41,7 +41,7 @@ type BufferManager interface {
 	DropChannel(channel string)
 	DropPartitions(channel string, partitionIDs []int64)
 	// BufferData put data into channel write buffer.
-	BufferData(channel string, insertData []*InsertData, deleteMsgs []*msgstream.DeleteMsg, startPos, endPos *msgpb.MsgPosition) error
+	BufferData(channel string, insertData []*InsertData, deleteMsgs []*msgstream.DeleteMsg, startPos, endPos *msgpb.MsgPosition, schemaVersion int32) error
 	// GetCheckpoint returns checkpoint for provided channel.
 	GetCheckpoint(channel string) (*msgpb.MsgPosition, bool, error)
 	// NotifyCheckpointUpdated notify write buffer checkpoint updated to reset flushTs.
@@ -175,7 +175,7 @@ func (m *bufferManager) Register(channel string, metacache metacache.MetaCache,
 }

 // CreateNewGrowingSegment notifies writeBuffer to create a new growing segment.
-func (m *bufferManager) CreateNewGrowingSegment(ctx context.Context, channel string, partitionID int64, segmentID int64) error {
+func (m *bufferManager) CreateNewGrowingSegment(ctx context.Context, channel string, partitionID int64, segmentID int64, schemaVersion int32) error {
 	buf, loaded := m.buffers.Get(channel)
 	if !loaded {
 		log.Ctx(ctx).Warn("write buffer not found when create new growing segment",
@@ -184,7 +184,7 @@ func (m *bufferManager) CreateNewGrowingSegment(ctx context.Context, channel str
 			zap.Int64("segmentID", segmentID))
 		return merr.WrapErrChannelNotFound(channel)
 	}
-	buf.CreateNewGrowingSegment(partitionID, segmentID, nil)
+	buf.CreateNewGrowingSegment(partitionID, segmentID, nil, schemaVersion)
 	return nil
 }

@@ -227,7 +227,7 @@ func (m *bufferManager) FlushChannel(ctx context.Context, channel string, flushT
 }

 // BufferData put data into channel write buffer.
-func (m *bufferManager) BufferData(channel string, insertData []*InsertData, deleteMsgs []*msgstream.DeleteMsg, startPos, endPos *msgpb.MsgPosition) error {
+func (m *bufferManager) BufferData(channel string, insertData []*InsertData, deleteMsgs []*msgstream.DeleteMsg, startPos, endPos *msgpb.MsgPosition, schemaVersion int32) error {
 	buf, loaded := m.buffers.Get(channel)
 	if !loaded {
 		log.Ctx(context.Background()).Warn("write buffer not found when buffer data",
@@ -235,7 +235,7 @@ func (m *bufferManager) BufferData(channel string, insertData []*InsertData, del
 		return merr.WrapErrChannelNotFound(channel)
 	}

-	return buf.BufferData(insertData, deleteMsgs, startPos, endPos)
+	return buf.BufferData(insertData, deleteMsgs, startPos, endPos, schemaVersion)
 }

 // GetCheckpoint returns checkpoint for provided channel.
@@ -16,6 +16,7 @@ import (
 	"github.com/milvus-io/milvus/internal/flushcommon/metacache"
 	"github.com/milvus-io/milvus/internal/flushcommon/syncmgr"
 	"github.com/milvus-io/milvus/pkg/v2/common"
+	"github.com/milvus-io/milvus/pkg/v2/proto/datapb"
 	"github.com/milvus-io/milvus/pkg/v2/util/hardware"
 	"github.com/milvus-io/milvus/pkg/v2/util/merr"
 	"github.com/milvus-io/milvus/pkg/v2/util/paramtable"
@@ -142,11 +143,13 @@ func (s *ManagerSuite) TestFlushAllSegments() {

 func (s *ManagerSuite) TestCreateNewGrowingSegment() {
 	manager := s.manager
-	err := manager.CreateNewGrowingSegment(context.Background(), s.channelName, 1, 1)
+	err := manager.CreateNewGrowingSegment(context.Background(), s.channelName, 1, 1, 0)
 	s.Error(err)

 	s.metacache.EXPECT().GetSegmentByID(mock.Anything).Return(nil, false).Once()
-	s.metacache.EXPECT().AddSegment(mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return()
+	s.metacache.EXPECT().AddSegment(mock.MatchedBy(func(info *datapb.SegmentInfo) bool {
+		return info.GetSchemaVersion() == 100
+	}), mock.Anything, mock.Anything, mock.Anything).Return()

 	wb, err := NewL0WriteBuffer(s.channelName, s.metacache, s.syncMgr, &writeBufferOption{
 		idAllocator: s.allocator,
@@ -154,14 +157,14 @@ func (s *ManagerSuite) TestCreateNewGrowingSegment() {
 	s.NoError(err)

 	s.manager.buffers.Insert(s.channelName, wb)
-	err = manager.CreateNewGrowingSegment(context.Background(), s.channelName, 1, 1)
+	err = manager.CreateNewGrowingSegment(context.Background(), s.channelName, 1, 1, 100)
 	s.NoError(err)
 }

 func (s *ManagerSuite) TestBufferData() {
 	manager := s.manager
 	s.Run("channel_not_found", func() {
-		err := manager.BufferData(s.channelName, nil, nil, nil, nil)
+		err := manager.BufferData(s.channelName, nil, nil, nil, nil, 0)
 		s.Error(err, "BufferData shall return error when channel not found")
 	})

@@ -169,9 +172,9 @@ func (s *ManagerSuite) TestBufferData() {
 		wb := NewMockWriteBuffer(s.T())

 		s.manager.buffers.Insert(s.channelName, wb)
-		wb.EXPECT().BufferData(mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(nil)
+		wb.EXPECT().BufferData(mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(nil)

-		err := manager.BufferData(s.channelName, nil, nil, nil, nil)
+		err := manager.BufferData(s.channelName, nil, nil, nil, nil, 100)
 		s.NoError(err)
 	})
 }
@@ -26,17 +26,17 @@ func (_m *MockBufferManager) EXPECT() *MockBufferManager_Expecter {
 	return &MockBufferManager_Expecter{mock: &_m.Mock}
 }

-// BufferData provides a mock function with given fields: channel, insertData, deleteMsgs, startPos, endPos
-func (_m *MockBufferManager) BufferData(channel string, insertData []*InsertData, deleteMsgs []*msgstream.DeleteMsg, startPos *msgpb.MsgPosition, endPos *msgpb.MsgPosition) error {
-	ret := _m.Called(channel, insertData, deleteMsgs, startPos, endPos)
+// BufferData provides a mock function with given fields: channel, insertData, deleteMsgs, startPos, endPos, schemaVersion
+func (_m *MockBufferManager) BufferData(channel string, insertData []*InsertData, deleteMsgs []*msgstream.DeleteMsg, startPos *msgpb.MsgPosition, endPos *msgpb.MsgPosition, schemaVersion int32) error {
+	ret := _m.Called(channel, insertData, deleteMsgs, startPos, endPos, schemaVersion)

 	if len(ret) == 0 {
 		panic("no return value specified for BufferData")
 	}

 	var r0 error
-	if rf, ok := ret.Get(0).(func(string, []*InsertData, []*msgstream.DeleteMsg, *msgpb.MsgPosition, *msgpb.MsgPosition) error); ok {
-		r0 = rf(channel, insertData, deleteMsgs, startPos, endPos)
+	if rf, ok := ret.Get(0).(func(string, []*InsertData, []*msgstream.DeleteMsg, *msgpb.MsgPosition, *msgpb.MsgPosition, int32) error); ok {
+		r0 = rf(channel, insertData, deleteMsgs, startPos, endPos, schemaVersion)
 	} else {
 		r0 = ret.Error(0)
 	}
@@ -55,13 +55,14 @@ type MockBufferManager_BufferData_Call struct {
 //   - deleteMsgs []*msgstream.DeleteMsg
 //   - startPos *msgpb.MsgPosition
 //   - endPos *msgpb.MsgPosition
-func (_e *MockBufferManager_Expecter) BufferData(channel interface{}, insertData interface{}, deleteMsgs interface{}, startPos interface{}, endPos interface{}) *MockBufferManager_BufferData_Call {
-	return &MockBufferManager_BufferData_Call{Call: _e.mock.On("BufferData", channel, insertData, deleteMsgs, startPos, endPos)}
+//   - schemaVersion int32
+func (_e *MockBufferManager_Expecter) BufferData(channel interface{}, insertData interface{}, deleteMsgs interface{}, startPos interface{}, endPos interface{}, schemaVersion interface{}) *MockBufferManager_BufferData_Call {
+	return &MockBufferManager_BufferData_Call{Call: _e.mock.On("BufferData", channel, insertData, deleteMsgs, startPos, endPos, schemaVersion)}
 }

-func (_c *MockBufferManager_BufferData_Call) Run(run func(channel string, insertData []*InsertData, deleteMsgs []*msgstream.DeleteMsg, startPos *msgpb.MsgPosition, endPos *msgpb.MsgPosition)) *MockBufferManager_BufferData_Call {
+func (_c *MockBufferManager_BufferData_Call) Run(run func(channel string, insertData []*InsertData, deleteMsgs []*msgstream.DeleteMsg, startPos *msgpb.MsgPosition, endPos *msgpb.MsgPosition, schemaVersion int32)) *MockBufferManager_BufferData_Call {
 	_c.Call.Run(func(args mock.Arguments) {
-		run(args[0].(string), args[1].([]*InsertData), args[2].([]*msgstream.DeleteMsg), args[3].(*msgpb.MsgPosition), args[4].(*msgpb.MsgPosition))
+		run(args[0].(string), args[1].([]*InsertData), args[2].([]*msgstream.DeleteMsg), args[3].(*msgpb.MsgPosition), args[4].(*msgpb.MsgPosition), args[5].(int32))
 	})
 	return _c
 }
@@ -71,22 +72,22 @@ func (_c *MockBufferManager_BufferData_Call) Return(_a0 error) *MockBufferManage
 	return _c
 }

-func (_c *MockBufferManager_BufferData_Call) RunAndReturn(run func(string, []*InsertData, []*msgstream.DeleteMsg, *msgpb.MsgPosition, *msgpb.MsgPosition) error) *MockBufferManager_BufferData_Call {
+func (_c *MockBufferManager_BufferData_Call) RunAndReturn(run func(string, []*InsertData, []*msgstream.DeleteMsg, *msgpb.MsgPosition, *msgpb.MsgPosition, int32) error) *MockBufferManager_BufferData_Call {
 	_c.Call.Return(run)
 	return _c
 }

-// CreateNewGrowingSegment provides a mock function with given fields: ctx, channel, partition, segmentID
-func (_m *MockBufferManager) CreateNewGrowingSegment(ctx context.Context, channel string, partition int64, segmentID int64) error {
-	ret := _m.Called(ctx, channel, partition, segmentID)
+// CreateNewGrowingSegment provides a mock function with given fields: ctx, channel, partition, segmentID, schemaVersion
+func (_m *MockBufferManager) CreateNewGrowingSegment(ctx context.Context, channel string, partition int64, segmentID int64, schemaVersion int32) error {
+	ret := _m.Called(ctx, channel, partition, segmentID, schemaVersion)

 	if len(ret) == 0 {
 		panic("no return value specified for CreateNewGrowingSegment")
 	}

 	var r0 error
-	if rf, ok := ret.Get(0).(func(context.Context, string, int64, int64) error); ok {
-		r0 = rf(ctx, channel, partition, segmentID)
+	if rf, ok := ret.Get(0).(func(context.Context, string, int64, int64, int32) error); ok {
+		r0 = rf(ctx, channel, partition, segmentID, schemaVersion)
 	} else {
 		r0 = ret.Error(0)
 	}
@@ -104,13 +105,14 @@ type MockBufferManager_CreateNewGrowingSegment_Call struct {
 //   - channel string
 //   - partition int64
 //   - segmentID int64
-func (_e *MockBufferManager_Expecter) CreateNewGrowingSegment(ctx interface{}, channel interface{}, partition interface{}, segmentID interface{}) *MockBufferManager_CreateNewGrowingSegment_Call {
-	return &MockBufferManager_CreateNewGrowingSegment_Call{Call: _e.mock.On("CreateNewGrowingSegment", ctx, channel, partition, segmentID)}
+//   - schemaVersion int32
+func (_e *MockBufferManager_Expecter) CreateNewGrowingSegment(ctx interface{}, channel interface{}, partition interface{}, segmentID interface{}, schemaVersion interface{}) *MockBufferManager_CreateNewGrowingSegment_Call {
+	return &MockBufferManager_CreateNewGrowingSegment_Call{Call: _e.mock.On("CreateNewGrowingSegment", ctx, channel, partition, segmentID, schemaVersion)}
 }

-func (_c *MockBufferManager_CreateNewGrowingSegment_Call) Run(run func(ctx context.Context, channel string, partition int64, segmentID int64)) *MockBufferManager_CreateNewGrowingSegment_Call {
+func (_c *MockBufferManager_CreateNewGrowingSegment_Call) Run(run func(ctx context.Context, channel string, partition int64, segmentID int64, schemaVersion int32)) *MockBufferManager_CreateNewGrowingSegment_Call {
 	_c.Call.Run(func(args mock.Arguments) {
-		run(args[0].(context.Context), args[1].(string), args[2].(int64), args[3].(int64))
+		run(args[0].(context.Context), args[1].(string), args[2].(int64), args[3].(int64), args[4].(int32))
 	})
 	return _c
 }
@@ -120,7 +122,7 @@ func (_c *MockBufferManager_CreateNewGrowingSegment_Call) Return(_a0 error) *Moc
 	return _c
 }

-func (_c *MockBufferManager_CreateNewGrowingSegment_Call) RunAndReturn(run func(context.Context, string, int64, int64) error) *MockBufferManager_CreateNewGrowingSegment_Call {
+func (_c *MockBufferManager_CreateNewGrowingSegment_Call) RunAndReturn(run func(context.Context, string, int64, int64, int32) error) *MockBufferManager_CreateNewGrowingSegment_Call {
 	_c.Call.Return(run)
 	return _c
 }
@@ -24,17 +24,17 @@ func (_m *MockWriteBuffer) EXPECT() *MockWriteBuffer_Expecter {
 	return &MockWriteBuffer_Expecter{mock: &_m.Mock}
 }

-// BufferData provides a mock function with given fields: insertMsgs, deleteMsgs, startPos, endPos
-func (_m *MockWriteBuffer) BufferData(insertMsgs []*InsertData, deleteMsgs []*msgstream.DeleteMsg, startPos *msgpb.MsgPosition, endPos *msgpb.MsgPosition) error {
-	ret := _m.Called(insertMsgs, deleteMsgs, startPos, endPos)
+// BufferData provides a mock function with given fields: insertMsgs, deleteMsgs, startPos, endPos, schemaVersion
+func (_m *MockWriteBuffer) BufferData(insertMsgs []*InsertData, deleteMsgs []*msgstream.DeleteMsg, startPos *msgpb.MsgPosition, endPos *msgpb.MsgPosition, schemaVersion int32) error {
+	ret := _m.Called(insertMsgs, deleteMsgs, startPos, endPos, schemaVersion)

 	if len(ret) == 0 {
 		panic("no return value specified for BufferData")
 	}

 	var r0 error
-	if rf, ok := ret.Get(0).(func([]*InsertData, []*msgstream.DeleteMsg, *msgpb.MsgPosition, *msgpb.MsgPosition) error); ok {
-		r0 = rf(insertMsgs, deleteMsgs, startPos, endPos)
+	if rf, ok := ret.Get(0).(func([]*InsertData, []*msgstream.DeleteMsg, *msgpb.MsgPosition, *msgpb.MsgPosition, int32) error); ok {
+		r0 = rf(insertMsgs, deleteMsgs, startPos, endPos, schemaVersion)
 	} else {
 		r0 = ret.Error(0)
 	}
@@ -52,13 +52,14 @@ type MockWriteBuffer_BufferData_Call struct {
 //   - deleteMsgs []*msgstream.DeleteMsg
 //   - startPos *msgpb.MsgPosition
 //   - endPos *msgpb.MsgPosition
-func (_e *MockWriteBuffer_Expecter) BufferData(insertMsgs interface{}, deleteMsgs interface{}, startPos interface{}, endPos interface{}) *MockWriteBuffer_BufferData_Call {
-	return &MockWriteBuffer_BufferData_Call{Call: _e.mock.On("BufferData", insertMsgs, deleteMsgs, startPos, endPos)}
+//   - schemaVersion int32
+func (_e *MockWriteBuffer_Expecter) BufferData(insertMsgs interface{}, deleteMsgs interface{}, startPos interface{}, endPos interface{}, schemaVersion interface{}) *MockWriteBuffer_BufferData_Call {
+	return &MockWriteBuffer_BufferData_Call{Call: _e.mock.On("BufferData", insertMsgs, deleteMsgs, startPos, endPos, schemaVersion)}
 }

-func (_c *MockWriteBuffer_BufferData_Call) Run(run func(insertMsgs []*InsertData, deleteMsgs []*msgstream.DeleteMsg, startPos *msgpb.MsgPosition, endPos *msgpb.MsgPosition)) *MockWriteBuffer_BufferData_Call {
+func (_c *MockWriteBuffer_BufferData_Call) Run(run func(insertMsgs []*InsertData, deleteMsgs []*msgstream.DeleteMsg, startPos *msgpb.MsgPosition, endPos *msgpb.MsgPosition, schemaVersion int32)) *MockWriteBuffer_BufferData_Call {
 	_c.Call.Run(func(args mock.Arguments) {
-		run(args[0].([]*InsertData), args[1].([]*msgstream.DeleteMsg), args[2].(*msgpb.MsgPosition), args[3].(*msgpb.MsgPosition))
+		run(args[0].([]*InsertData), args[1].([]*msgstream.DeleteMsg), args[2].(*msgpb.MsgPosition), args[3].(*msgpb.MsgPosition), args[4].(int32))
 	})
 	return _c
 }
@@ -68,7 +69,7 @@ func (_c *MockWriteBuffer_BufferData_Call) Return(_a0 error) *MockWriteBuffer_Bu
 	return _c
 }

-func (_c *MockWriteBuffer_BufferData_Call) RunAndReturn(run func([]*InsertData, []*msgstream.DeleteMsg, *msgpb.MsgPosition, *msgpb.MsgPosition) error) *MockWriteBuffer_BufferData_Call {
+func (_c *MockWriteBuffer_BufferData_Call) RunAndReturn(run func([]*InsertData, []*msgstream.DeleteMsg, *msgpb.MsgPosition, *msgpb.MsgPosition, int32) error) *MockWriteBuffer_BufferData_Call {
 	_c.Call.Return(run)
 	return _c
 }
@@ -107,9 +108,9 @@ func (_c *MockWriteBuffer_Close_Call) RunAndReturn(run func(context.Context, boo
 	return _c
 }

-// CreateNewGrowingSegment provides a mock function with given fields: partitionID, segmentID, startPos
-func (_m *MockWriteBuffer) CreateNewGrowingSegment(partitionID int64, segmentID int64, startPos *msgpb.MsgPosition) {
-	_m.Called(partitionID, segmentID, startPos)
+// CreateNewGrowingSegment provides a mock function with given fields: partitionID, segmentID, startPos, schemaVersion
+func (_m *MockWriteBuffer) CreateNewGrowingSegment(partitionID int64, segmentID int64, startPos *msgpb.MsgPosition, schemaVersion int32) {
+	_m.Called(partitionID, segmentID, startPos, schemaVersion)
 }

 // MockWriteBuffer_CreateNewGrowingSegment_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'CreateNewGrowingSegment'
@@ -121,13 +122,14 @@ type MockWriteBuffer_CreateNewGrowingSegment_Call struct {
 //   - partitionID int64
 //   - segmentID int64
 //   - startPos *msgpb.MsgPosition
-func (_e *MockWriteBuffer_Expecter) CreateNewGrowingSegment(partitionID interface{}, segmentID interface{}, startPos interface{}) *MockWriteBuffer_CreateNewGrowingSegment_Call {
-	return &MockWriteBuffer_CreateNewGrowingSegment_Call{Call: _e.mock.On("CreateNewGrowingSegment", partitionID, segmentID, startPos)}
+//   - schemaVersion int32
+func (_e *MockWriteBuffer_Expecter) CreateNewGrowingSegment(partitionID interface{}, segmentID interface{}, startPos interface{}, schemaVersion interface{}) *MockWriteBuffer_CreateNewGrowingSegment_Call {
+	return &MockWriteBuffer_CreateNewGrowingSegment_Call{Call: _e.mock.On("CreateNewGrowingSegment", partitionID, segmentID, startPos, schemaVersion)}
 }

-func (_c *MockWriteBuffer_CreateNewGrowingSegment_Call) Run(run func(partitionID int64, segmentID int64, startPos *msgpb.MsgPosition)) *MockWriteBuffer_CreateNewGrowingSegment_Call {
+func (_c *MockWriteBuffer_CreateNewGrowingSegment_Call) Run(run func(partitionID int64, segmentID int64, startPos *msgpb.MsgPosition, schemaVersion int32)) *MockWriteBuffer_CreateNewGrowingSegment_Call {
 	_c.Call.Run(func(args mock.Arguments) {
-		run(args[0].(int64), args[1].(int64), args[2].(*msgpb.MsgPosition))
+		run(args[0].(int64), args[1].(int64), args[2].(*msgpb.MsgPosition), args[3].(int32))
 	})
 	return _c
 }
@@ -137,7 +139,7 @@ func (_c *MockWriteBuffer_CreateNewGrowingSegment_Call) Return() *MockWriteBuffe
 	return _c
 }

-func (_c *MockWriteBuffer_CreateNewGrowingSegment_Call) RunAndReturn(run func(int64, int64, *msgpb.MsgPosition)) *MockWriteBuffer_CreateNewGrowingSegment_Call {
+func (_c *MockWriteBuffer_CreateNewGrowingSegment_Call) RunAndReturn(run func(int64, int64, *msgpb.MsgPosition, int32)) *MockWriteBuffer_CreateNewGrowingSegment_Call {
 	_c.Run(run)
 	return _c
 }
@@ -44,9 +44,9 @@ type WriteBuffer interface {
 	// HasSegment checks whether certain segment exists in this buffer.
 	HasSegment(segmentID int64) bool
 	// CreateNewGrowingSegment creates a new growing segment in the buffer.
-	CreateNewGrowingSegment(partitionID int64, segmentID int64, startPos *msgpb.MsgPosition)
+	CreateNewGrowingSegment(partitionID int64, segmentID int64, startPos *msgpb.MsgPosition, schemaVersion int32)
 	// BufferData is the method to buffer dml data msgs.
-	BufferData(insertMsgs []*InsertData, deleteMsgs []*msgstream.DeleteMsg, startPos, endPos *msgpb.MsgPosition) error
+	BufferData(insertMsgs []*InsertData, deleteMsgs []*msgstream.DeleteMsg, startPos, endPos *msgpb.MsgPosition, schemaVersion int32) error
 	// FlushTimestamp set flush timestamp for write buffer
 	SetFlushTimestamp(flushTs uint64)
 	// GetFlushTimestamp get current flush timestamp
@@ -522,7 +522,7 @@ func (id *InsertData) batchPkExists(pks []storage.PrimaryKey, tss []uint64, hits
 	return hits
 }

-func (wb *writeBufferBase) CreateNewGrowingSegment(partitionID int64, segmentID int64, startPos *msgpb.MsgPosition) {
+func (wb *writeBufferBase) CreateNewGrowingSegment(partitionID int64, segmentID int64, startPos *msgpb.MsgPosition, schemaVersion int32) {
 	_, ok := wb.metaCache.GetSegmentByID(segmentID)
 	// new segment
 	if !ok {
@@ -545,6 +545,7 @@ func (wb *writeBufferBase) CreateNewGrowingSegment(partitionID int64, segmentID
 			State:          commonpb.SegmentState_Growing,
 			StorageVersion: storageVersion,
 			ManifestPath:   manifestPath,
+			SchemaVersion:  schemaVersion,
 		}
 		wb.metaCache.AddSegment(segmentInfo, func(_ *datapb.SegmentInfo) pkoracle.PkStat {
 			return pkoracle.NewBloomFilterSetWithBatchSize(wb.getEstBatchSize())
@@ -5,7 +5,9 @@ package mock_util
 import (
 	context "context"

+	messagespb "github.com/milvus-io/milvus/pkg/v2/proto/messagespb"
 	message "github.com/milvus-io/milvus/pkg/v2/streaming/util/message"
+
 	mock "github.com/stretchr/testify/mock"
 )

@@ -23,7 +25,7 @@ func (_m *MockMsgHandler) EXPECT() *MockMsgHandler_Expecter {
 }

 // HandleAlterCollection provides a mock function with given fields: ctx, alterCollectionMsg
-func (_m *MockMsgHandler) HandleAlterCollection(ctx context.Context, alterCollectionMsg message.ImmutableAlterCollectionMessageV2) error {
+func (_m *MockMsgHandler) HandleAlterCollection(ctx context.Context, alterCollectionMsg message.SpecializedImmutableMessage[*messagespb.AlterCollectionMessageHeader, *messagespb.AlterCollectionMessageBody]) error {
 	ret := _m.Called(ctx, alterCollectionMsg)

 	if len(ret) == 0 {
@@ -31,7 +33,7 @@ func (_m *MockMsgHandler) HandleAlterCollection(ctx context.Context, alterCollec
 	}

 	var r0 error
-	if rf, ok := ret.Get(0).(func(context.Context, message.ImmutableAlterCollectionMessageV2) error); ok {
+	if rf, ok := ret.Get(0).(func(context.Context, message.SpecializedImmutableMessage[*messagespb.AlterCollectionMessageHeader, *messagespb.AlterCollectionMessageBody]) error); ok {
 		r0 = rf(ctx, alterCollectionMsg)
 	} else {
 		r0 = ret.Error(0)
@@ -47,14 +49,14 @@ type MockMsgHandler_HandleAlterCollection_Call struct {

 // HandleAlterCollection is a helper method to define mock.On call
 //   - ctx context.Context
-//   - alterCollectionMsg message.ImmutableAlterCollectionMessageV2
+//   - alterCollectionMsg message.SpecializedImmutableMessage[*messagespb.AlterCollectionMessageHeader,*messagespb.AlterCollectionMessageBody]
 func (_e *MockMsgHandler_Expecter) HandleAlterCollection(ctx interface{}, alterCollectionMsg interface{}) *MockMsgHandler_HandleAlterCollection_Call {
 	return &MockMsgHandler_HandleAlterCollection_Call{Call: _e.mock.On("HandleAlterCollection", ctx, alterCollectionMsg)}
 }

-func (_c *MockMsgHandler_HandleAlterCollection_Call) Run(run func(ctx context.Context, alterCollectionMsg message.ImmutableAlterCollectionMessageV2)) *MockMsgHandler_HandleAlterCollection_Call {
+func (_c *MockMsgHandler_HandleAlterCollection_Call) Run(run func(ctx context.Context, alterCollectionMsg message.SpecializedImmutableMessage[*messagespb.AlterCollectionMessageHeader, *messagespb.AlterCollectionMessageBody])) *MockMsgHandler_HandleAlterCollection_Call {
 	_c.Call.Run(func(args mock.Arguments) {
-		run(args[0].(context.Context), args[1].(message.ImmutableAlterCollectionMessageV2))
+		run(args[0].(context.Context), args[1].(message.SpecializedImmutableMessage[*messagespb.AlterCollectionMessageHeader, *messagespb.AlterCollectionMessageBody]))
 	})
 	return _c
 }
@@ -64,7 +66,7 @@ func (_c *MockMsgHandler_HandleAlterCollection_Call) Return(_a0 error) *MockMsgH
 	return _c
 }

-func (_c *MockMsgHandler_HandleAlterCollection_Call) RunAndReturn(run func(context.Context, message.ImmutableAlterCollectionMessageV2) error) *MockMsgHandler_HandleAlterCollection_Call {
+func (_c *MockMsgHandler_HandleAlterCollection_Call) RunAndReturn(run func(context.Context, message.SpecializedImmutableMessage[*messagespb.AlterCollectionMessageHeader, *messagespb.AlterCollectionMessageBody]) error) *MockMsgHandler_HandleAlterCollection_Call {
 	_c.Call.Return(run)
 	return _c
 }
@@ -118,7 +120,7 @@ func (_c *MockMsgHandler_HandleAlterWAL_Call) RunAndReturn(run func(context.Cont
 }

 // HandleCreateSegment provides a mock function with given fields: ctx, createSegmentMsg
-func (_m *MockMsgHandler) HandleCreateSegment(ctx context.Context, createSegmentMsg message.ImmutableCreateSegmentMessageV2) error {
+func (_m *MockMsgHandler) HandleCreateSegment(ctx context.Context, createSegmentMsg message.SpecializedImmutableMessage[*messagespb.CreateSegmentMessageHeader, *messagespb.CreateSegmentMessageBody]) error {
 	ret := _m.Called(ctx, createSegmentMsg)

 	if len(ret) == 0 {
@@ -126,7 +128,7 @@ func (_m *MockMsgHandler) HandleCreateSegment(ctx context.Context, createSegment
 	}

 	var r0 error
-	if rf, ok := ret.Get(0).(func(context.Context, message.ImmutableCreateSegmentMessageV2) error); ok {
+	if rf, ok := ret.Get(0).(func(context.Context, message.SpecializedImmutableMessage[*messagespb.CreateSegmentMessageHeader, *messagespb.CreateSegmentMessageBody]) error); ok {
 		r0 = rf(ctx, createSegmentMsg)
 	} else {
 		r0 = ret.Error(0)
@@ -142,14 +144,14 @@ type MockMsgHandler_HandleCreateSegment_Call struct {

 // HandleCreateSegment is a helper method to define mock.On call
 //   - ctx context.Context
-//   - createSegmentMsg message.ImmutableCreateSegmentMessageV2
+//   - createSegmentMsg message.SpecializedImmutableMessage[*messagespb.CreateSegmentMessageHeader,*messagespb.CreateSegmentMessageBody]
 func (_e *MockMsgHandler_Expecter) HandleCreateSegment(ctx interface{}, createSegmentMsg interface{}) *MockMsgHandler_HandleCreateSegment_Call {
 	return &MockMsgHandler_HandleCreateSegment_Call{Call: _e.mock.On("HandleCreateSegment", ctx, createSegmentMsg)}
 }

-func (_c *MockMsgHandler_HandleCreateSegment_Call) Run(run func(ctx context.Context, createSegmentMsg message.ImmutableCreateSegmentMessageV2)) *MockMsgHandler_HandleCreateSegment_Call {
+func (_c *MockMsgHandler_HandleCreateSegment_Call) Run(run func(ctx context.Context, createSegmentMsg message.SpecializedImmutableMessage[*messagespb.CreateSegmentMessageHeader, *messagespb.CreateSegmentMessageBody])) *MockMsgHandler_HandleCreateSegment_Call {
 	_c.Call.Run(func(args mock.Arguments) {
-		run(args[0].(context.Context), args[1].(message.ImmutableCreateSegmentMessageV2))
+		run(args[0].(context.Context), args[1].(message.SpecializedImmutableMessage[*messagespb.CreateSegmentMessageHeader, *messagespb.CreateSegmentMessageBody]))
 	})
 	return _c
 }
@@ -159,13 +161,13 @@ func (_c *MockMsgHandler_HandleCreateSegment_Call) Return(_a0 error) *MockMsgHan
 	return _c
 }

-func (_c *MockMsgHandler_HandleCreateSegment_Call) RunAndReturn(run func(context.Context, message.ImmutableCreateSegmentMessageV2) error) *MockMsgHandler_HandleCreateSegment_Call {
+func (_c *MockMsgHandler_HandleCreateSegment_Call) RunAndReturn(run func(context.Context, message.SpecializedImmutableMessage[*messagespb.CreateSegmentMessageHeader, *messagespb.CreateSegmentMessageBody]) error) *MockMsgHandler_HandleCreateSegment_Call {
 	_c.Call.Return(run)
 	return _c
 }

 // HandleFlush provides a mock function with given fields: flushMsg
-func (_m *MockMsgHandler) HandleFlush(flushMsg message.ImmutableFlushMessageV2) error {
+func (_m *MockMsgHandler) HandleFlush(flushMsg message.SpecializedImmutableMessage[*messagespb.FlushMessageHeader, *messagespb.FlushMessageBody]) error {
 	ret := _m.Called(flushMsg)

 	if len(ret) == 0 {
@@ -173,7 +175,7 @@ func (_m *MockMsgHandler) HandleFlush(flushMsg message.ImmutableFlushMessageV2)
 	}

 	var r0 error
-	if rf, ok := ret.Get(0).(func(message.ImmutableFlushMessageV2) error); ok {
+	if rf, ok := ret.Get(0).(func(message.SpecializedImmutableMessage[*messagespb.FlushMessageHeader, *messagespb.FlushMessageBody]) error); ok {
 		r0 = rf(flushMsg)
 	} else {
 		r0 = ret.Error(0)
@@ -188,14 +190,14 @@ type MockMsgHandler_HandleFlush_Call struct {
 }

 // HandleFlush is a helper method to define mock.On call
-//   - flushMsg message.ImmutableFlushMessageV2
+//   - flushMsg message.SpecializedImmutableMessage[*messagespb.FlushMessageHeader,*messagespb.FlushMessageBody]
 func (_e *MockMsgHandler_Expecter) HandleFlush(flushMsg interface{}) *MockMsgHandler_HandleFlush_Call {
 	return &MockMsgHandler_HandleFlush_Call{Call: _e.mock.On("HandleFlush", flushMsg)}
 }

-func (_c *MockMsgHandler_HandleFlush_Call) Run(run func(flushMsg message.ImmutableFlushMessageV2)) *MockMsgHandler_HandleFlush_Call {
+func (_c *MockMsgHandler_HandleFlush_Call) Run(run func(flushMsg message.SpecializedImmutableMessage[*messagespb.FlushMessageHeader, *messagespb.FlushMessageBody])) *MockMsgHandler_HandleFlush_Call {
 	_c.Call.Run(func(args mock.Arguments) {
-		run(args[0].(message.ImmutableFlushMessageV2))
+		run(args[0].(message.SpecializedImmutableMessage[*messagespb.FlushMessageHeader, *messagespb.FlushMessageBody]))
 	})
 	return _c
 }
@@ -205,13 +207,13 @@ func (_c *MockMsgHandler_HandleFlush_Call) Return(_a0 error) *MockMsgHandler_Han
 	return _c
 }

-func (_c *MockMsgHandler_HandleFlush_Call) RunAndReturn(run func(message.ImmutableFlushMessageV2) error) *MockMsgHandler_HandleFlush_Call {
+func (_c *MockMsgHandler_HandleFlush_Call) RunAndReturn(run func(message.SpecializedImmutableMessage[*messagespb.FlushMessageHeader, *messagespb.FlushMessageBody]) error) *MockMsgHandler_HandleFlush_Call {
 	_c.Call.Return(run)
 	return _c
 }

 // HandleFlushAll provides a mock function with given fields: vchannel, flushAllMsg
-func (_m *MockMsgHandler) HandleFlushAll(vchannel string, flushAllMsg message.ImmutableFlushAllMessageV2) error {
+func (_m *MockMsgHandler) HandleFlushAll(vchannel string, flushAllMsg message.SpecializedImmutableMessage[*messagespb.FlushAllMessageHeader, *messagespb.FlushAllMessageBody]) error {
 	ret := _m.Called(vchannel, flushAllMsg)

 	if len(ret) == 0 {
@@ -219,7 +221,7 @@ func (_m *MockMsgHandler) HandleFlushAll(vchannel string, flushAllMsg message.Im
 	}

 	var r0 error
-	if rf, ok := ret.Get(0).(func(string, message.ImmutableFlushAllMessageV2) error); ok {
+	if rf, ok := ret.Get(0).(func(string, message.SpecializedImmutableMessage[*messagespb.FlushAllMessageHeader, *messagespb.FlushAllMessageBody]) error); ok {
 		r0 = rf(vchannel, flushAllMsg)
 	} else {
 		r0 = ret.Error(0)
@@ -235,14 +237,14 @@ type MockMsgHandler_HandleFlushAll_Call struct {

 // HandleFlushAll is a helper method to define mock.On call
 //   - vchannel string
-//   - flushAllMsg message.ImmutableFlushAllMessageV2
+//   - flushAllMsg message.SpecializedImmutableMessage[*messagespb.FlushAllMessageHeader,*messagespb.FlushAllMessageBody]
 func (_e *MockMsgHandler_Expecter) HandleFlushAll(vchannel interface{}, flushAllMsg interface{}) *MockMsgHandler_HandleFlushAll_Call {
 	return &MockMsgHandler_HandleFlushAll_Call{Call: _e.mock.On("HandleFlushAll", vchannel, flushAllMsg)}
 }

-func (_c *MockMsgHandler_HandleFlushAll_Call) Run(run func(vchannel string, flushAllMsg message.ImmutableFlushAllMessageV2)) *MockMsgHandler_HandleFlushAll_Call {
+func (_c *MockMsgHandler_HandleFlushAll_Call) Run(run func(vchannel string, flushAllMsg message.SpecializedImmutableMessage[*messagespb.FlushAllMessageHeader, *messagespb.FlushAllMessageBody])) *MockMsgHandler_HandleFlushAll_Call {
 	_c.Call.Run(func(args mock.Arguments) {
-		run(args[0].(string), args[1].(message.ImmutableFlushAllMessageV2))
+		run(args[0].(string), args[1].(message.SpecializedImmutableMessage[*messagespb.FlushAllMessageHeader, *messagespb.FlushAllMessageBody]))
 	})
 	return _c
 }
@@ -252,13 +254,13 @@ func (_c *MockMsgHandler_HandleFlushAll_Call) Return(_a0 error) *MockMsgHandler_
 	return _c
 }

-func (_c *MockMsgHandler_HandleFlushAll_Call) RunAndReturn(run func(string, message.ImmutableFlushAllMessageV2) error) *MockMsgHandler_HandleFlushAll_Call {
+func (_c *MockMsgHandler_HandleFlushAll_Call) RunAndReturn(run func(string, message.SpecializedImmutableMessage[*messagespb.FlushAllMessageHeader, *messagespb.FlushAllMessageBody]) error) *MockMsgHandler_HandleFlushAll_Call {
 	_c.Call.Return(run)
 	return _c
 }

 // HandleManualFlush provides a mock function with given fields: flushMsg
-func (_m *MockMsgHandler) HandleManualFlush(flushMsg message.ImmutableManualFlushMessageV2) error {
+func (_m *MockMsgHandler) HandleManualFlush(flushMsg message.SpecializedImmutableMessage[*messagespb.ManualFlushMessageHeader, *messagespb.ManualFlushMessageBody]) error {
 	ret := _m.Called(flushMsg)

 	if len(ret) == 0 {
@@ -266,7 +268,7 @@ func (_m *MockMsgHandler) HandleManualFlush(flushMsg message.ImmutableManualFlus
 	}

 	var r0 error
-	if rf, ok := ret.Get(0).(func(message.ImmutableManualFlushMessageV2) error); ok {
+	if rf, ok := ret.Get(0).(func(message.SpecializedImmutableMessage[*messagespb.ManualFlushMessageHeader, *messagespb.ManualFlushMessageBody]) error); ok {
 		r0 = rf(flushMsg)
 	} else {
 		r0 = ret.Error(0)
@@ -281,14 +283,14 @@ type MockMsgHandler_HandleManualFlush_Call struct {
 }

 // HandleManualFlush is a helper method to define mock.On call
-//   - flushMsg message.ImmutableManualFlushMessageV2
+//   - flushMsg message.SpecializedImmutableMessage[*messagespb.ManualFlushMessageHeader,*messagespb.ManualFlushMessageBody]
 func (_e *MockMsgHandler_Expecter) HandleManualFlush(flushMsg interface{}) *MockMsgHandler_HandleManualFlush_Call {
 	return &MockMsgHandler_HandleManualFlush_Call{Call: _e.mock.On("HandleManualFlush", flushMsg)}
 }

-func (_c *MockMsgHandler_HandleManualFlush_Call) Run(run func(flushMsg message.ImmutableManualFlushMessageV2)) *MockMsgHandler_HandleManualFlush_Call {
+func (_c *MockMsgHandler_HandleManualFlush_Call) Run(run func(flushMsg message.SpecializedImmutableMessage[*messagespb.ManualFlushMessageHeader, *messagespb.ManualFlushMessageBody])) *MockMsgHandler_HandleManualFlush_Call {
 	_c.Call.Run(func(args mock.Arguments) {
-		run(args[0].(message.ImmutableManualFlushMessageV2))
+		run(args[0].(message.SpecializedImmutableMessage[*messagespb.ManualFlushMessageHeader, *messagespb.ManualFlushMessageBody]))
 	})
 	return _c
 }
@@ -298,13 +300,13 @@ func (_c *MockMsgHandler_HandleManualFlush_Call) Return(_a0 error) *MockMsgHandl
 	return _c
 }

-func (_c *MockMsgHandler_HandleManualFlush_Call) RunAndReturn(run func(message.ImmutableManualFlushMessageV2) error) *MockMsgHandler_HandleManualFlush_Call {
+func (_c *MockMsgHandler_HandleManualFlush_Call) RunAndReturn(run func(message.SpecializedImmutableMessage[*messagespb.ManualFlushMessageHeader, *messagespb.ManualFlushMessageBody]) error) *MockMsgHandler_HandleManualFlush_Call {
 	_c.Call.Return(run)
 	return _c
 }

 // HandleSchemaChange provides a mock function with given fields: ctx, schemaChangeMsg
-func (_m *MockMsgHandler) HandleSchemaChange(ctx context.Context, schemaChangeMsg message.ImmutableSchemaChangeMessageV2) error {
+func (_m *MockMsgHandler) HandleSchemaChange(ctx context.Context, schemaChangeMsg message.SpecializedImmutableMessage[*messagespb.SchemaChangeMessageHeader, *messagespb.SchemaChangeMessageBody]) error {
 	ret := _m.Called(ctx, schemaChangeMsg)

 	if len(ret) == 0 {
@@ -312,7 +314,7 @@ func (_m *MockMsgHandler) HandleSchemaChange(ctx context.Context, schemaChangeMs
 	}

 	var r0 error
-	if rf, ok := ret.Get(0).(func(context.Context, message.ImmutableSchemaChangeMessageV2) error); ok {
+	if rf, ok := ret.Get(0).(func(context.Context, message.SpecializedImmutableMessage[*messagespb.SchemaChangeMessageHeader, *messagespb.SchemaChangeMessageBody]) error); ok {
 		r0 = rf(ctx, schemaChangeMsg)
 	} else {
 		r0 = ret.Error(0)
@@ -328,14 +330,14 @@ type MockMsgHandler_HandleSchemaChange_Call struct {

 // HandleSchemaChange is a helper method to define mock.On call
 //   - ctx context.Context
-//   - schemaChangeMsg message.ImmutableSchemaChangeMessageV2
+//   - schemaChangeMsg message.SpecializedImmutableMessage[*messagespb.SchemaChangeMessageHeader,*messagespb.SchemaChangeMessageBody]
 func (_e *MockMsgHandler_Expecter) HandleSchemaChange(ctx interface{}, schemaChangeMsg interface{}) *MockMsgHandler_HandleSchemaChange_Call {
 	return &MockMsgHandler_HandleSchemaChange_Call{Call: _e.mock.On("HandleSchemaChange", ctx, schemaChangeMsg)}
 }

-func (_c *MockMsgHandler_HandleSchemaChange_Call) Run(run func(ctx context.Context, schemaChangeMsg message.ImmutableSchemaChangeMessageV2)) *MockMsgHandler_HandleSchemaChange_Call {
+func (_c *MockMsgHandler_HandleSchemaChange_Call) Run(run func(ctx context.Context, schemaChangeMsg message.SpecializedImmutableMessage[*messagespb.SchemaChangeMessageHeader, *messagespb.SchemaChangeMessageBody])) *MockMsgHandler_HandleSchemaChange_Call {
 	_c.Call.Run(func(args mock.Arguments) {
-		run(args[0].(context.Context), args[1].(message.ImmutableSchemaChangeMessageV2))
+		run(args[0].(context.Context), args[1].(message.SpecializedImmutableMessage[*messagespb.SchemaChangeMessageHeader, *messagespb.SchemaChangeMessageBody]))
 	})
 	return _c
 }
@@ -345,13 +347,13 @@ func (_c *MockMsgHandler_HandleSchemaChange_Call) Return(_a0 error) *MockMsgHand
 	return _c
 }

-func (_c *MockMsgHandler_HandleSchemaChange_Call) RunAndReturn(run func(context.Context, message.ImmutableSchemaChangeMessageV2) error) *MockMsgHandler_HandleSchemaChange_Call {
+func (_c *MockMsgHandler_HandleSchemaChange_Call) RunAndReturn(run func(context.Context, message.SpecializedImmutableMessage[*messagespb.SchemaChangeMessageHeader, *messagespb.SchemaChangeMessageBody]) error) *MockMsgHandler_HandleSchemaChange_Call {
 	_c.Call.Return(run)
 	return _c
 }

 // HandleTruncateCollection provides a mock function with given fields: truncateCollectionMsg
-func (_m *MockMsgHandler) HandleTruncateCollection(truncateCollectionMsg message.ImmutableTruncateCollectionMessageV2) error {
+func (_m *MockMsgHandler) HandleTruncateCollection(truncateCollectionMsg message.SpecializedImmutableMessage[*messagespb.TruncateCollectionMessageHeader, *messagespb.TruncateCollectionMessageBody]) error {
 	ret := _m.Called(truncateCollectionMsg)

 	if len(ret) == 0 {
@@ -359,7 +361,7 @@ func (_m *MockMsgHandler) HandleTruncateCollection(truncateCollectionMsg message
 	}

 	var r0 error
-	if rf, ok := ret.Get(0).(func(message.ImmutableTruncateCollectionMessageV2) error); ok {
+	if rf, ok := ret.Get(0).(func(message.SpecializedImmutableMessage[*messagespb.TruncateCollectionMessageHeader, *messagespb.TruncateCollectionMessageBody]) error); ok {
 		r0 = rf(truncateCollectionMsg)
 	} else {
 		r0 = ret.Error(0)
@@ -374,14 +376,14 @@ type MockMsgHandler_HandleTruncateCollection_Call struct {
 }

 // HandleTruncateCollection is a helper method to define mock.On call
-//   - truncateCollectionMsg message.ImmutableTruncateCollectionMessageV2
+//   - truncateCollectionMsg message.SpecializedImmutableMessage[*messagespb.TruncateCollectionMessageHeader,*messagespb.TruncateCollectionMessageBody]
 func (_e *MockMsgHandler_Expecter) HandleTruncateCollection(truncateCollectionMsg interface{}) *MockMsgHandler_HandleTruncateCollection_Call {
 	return &MockMsgHandler_HandleTruncateCollection_Call{Call: _e.mock.On("HandleTruncateCollection", truncateCollectionMsg)}
 }

-func (_c *MockMsgHandler_HandleTruncateCollection_Call) Run(run func(truncateCollectionMsg message.ImmutableTruncateCollectionMessageV2)) *MockMsgHandler_HandleTruncateCollection_Call {
+func (_c *MockMsgHandler_HandleTruncateCollection_Call) Run(run func(truncateCollectionMsg message.SpecializedImmutableMessage[*messagespb.TruncateCollectionMessageHeader, *messagespb.TruncateCollectionMessageBody])) *MockMsgHandler_HandleTruncateCollection_Call {
 	_c.Call.Run(func(args mock.Arguments) {
-		run(args[0].(message.ImmutableTruncateCollectionMessageV2))
+		run(args[0].(message.SpecializedImmutableMessage[*messagespb.TruncateCollectionMessageHeader, *messagespb.TruncateCollectionMessageBody]))
 	})
 	return _c
 }
@@ -391,7 +393,7 @@ func (_c *MockMsgHandler_HandleTruncateCollection_Call) Return(_a0 error) *MockM
 	return _c
 }

-func (_c *MockMsgHandler_HandleTruncateCollection_Call) RunAndReturn(run func(message.ImmutableTruncateCollectionMessageV2) error) *MockMsgHandler_HandleTruncateCollection_Call {
+func (_c *MockMsgHandler_HandleTruncateCollection_Call) RunAndReturn(run func(message.SpecializedImmutableMessage[*messagespb.TruncateCollectionMessageHeader, *messagespb.TruncateCollectionMessageBody]) error) *MockMsgHandler_HandleTruncateCollection_Call {
 	_c.Call.Return(run)
 	return _c
 }
@@ -27,6 +27,155 @@ func (_m *MockShardManager) EXPECT() *MockShardManager_Expecter {
 	return &MockShardManager_Expecter{mock: &_m.Mock}
 }

+// AppendNewCollectionSchema provides a mock function with given fields: msg
+func (_m *MockShardManager) AppendNewCollectionSchema(msg message.ImmutableAlterCollectionMessageV2) error {
+	ret := _m.Called(msg)
+
+	if len(ret) == 0 {
+		panic("no return value specified for AppendNewCollectionSchema")
+	}
+
+	var r0 error
+	if rf, ok := ret.Get(0).(func(message.ImmutableAlterCollectionMessageV2) error); ok {
+		r0 = rf(msg)
+	} else {
+		r0 = ret.Error(0)
+	}
+
+	return r0
+}
+
+// MockShardManager_AppendNewCollectionSchema_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'AppendNewCollectionSchema'
+type MockShardManager_AppendNewCollectionSchema_Call struct {
+	*mock.Call
+}
+
+// AppendNewCollectionSchema is a helper method to define mock.On call
+//   - msg message.ImmutableAlterCollectionMessageV2
+func (_e *MockShardManager_Expecter) AppendNewCollectionSchema(msg interface{}) *MockShardManager_AppendNewCollectionSchema_Call {
+	return &MockShardManager_AppendNewCollectionSchema_Call{Call: _e.mock.On("AppendNewCollectionSchema", msg)}
+}
+
+func (_c *MockShardManager_AppendNewCollectionSchema_Call) Run(run func(msg message.ImmutableAlterCollectionMessageV2)) *MockShardManager_AppendNewCollectionSchema_Call {
+	_c.Call.Run(func(args mock.Arguments) {
+		run(args[0].(message.ImmutableAlterCollectionMessageV2))
+	})
+	return _c
+}
+
+func (_c *MockShardManager_AppendNewCollectionSchema_Call) Return(_a0 error) *MockShardManager_AppendNewCollectionSchema_Call {
+	_c.Call.Return(_a0)
+	return _c
+}
+
+func (_c *MockShardManager_AppendNewCollectionSchema_Call) RunAndReturn(run func(message.ImmutableAlterCollectionMessageV2) error) *MockShardManager_AppendNewCollectionSchema_Call {
+	_c.Call.Return(run)
+	return _c
+}
+
+// AppendNewCollectionSchemaFromCreateCollection provides a mock function with given fields: msg
+func (_m *MockShardManager) AppendNewCollectionSchemaFromCreateCollection(msg message.ImmutableCreateCollectionMessageV1) error {
+	ret := _m.Called(msg)
+
+	if len(ret) == 0 {
+		panic("no return value specified for AppendNewCollectionSchemaFromCreateCollection")
+	}
+
+	var r0 error
+	if rf, ok := ret.Get(0).(func(message.ImmutableCreateCollectionMessageV1) error); ok {
+		r0 = rf(msg)
+	} else {
+		r0 = ret.Error(0)
+	}
+
+	return r0
+}
+
+// MockShardManager_AppendNewCollectionSchemaFromCreateCollection_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'AppendNewCollectionSchemaFromCreateCollection'
+type MockShardManager_AppendNewCollectionSchemaFromCreateCollection_Call struct {
+	*mock.Call
+}
+
+// AppendNewCollectionSchemaFromCreateCollection is a helper method to define mock.On call
+//   - msg message.ImmutableCreateCollectionMessageV1
+func (_e *MockShardManager_Expecter) AppendNewCollectionSchemaFromCreateCollection(msg interface{}) *MockShardManager_AppendNewCollectionSchemaFromCreateCollection_Call {
+	return &MockShardManager_AppendNewCollectionSchemaFromCreateCollection_Call{Call: _e.mock.On("AppendNewCollectionSchemaFromCreateCollection", msg)}
+}
+
+func (_c *MockShardManager_AppendNewCollectionSchemaFromCreateCollection_Call) Run(run func(msg message.ImmutableCreateCollectionMessageV1)) *MockShardManager_AppendNewCollectionSchemaFromCreateCollection_Call {
+	_c.Call.Run(func(args mock.Arguments) {
+		run(args[0].(message.ImmutableCreateCollectionMessageV1))
+	})
+	return _c
+}
+
+func (_c *MockShardManager_AppendNewCollectionSchemaFromCreateCollection_Call) Return(_a0 error) *MockShardManager_AppendNewCollectionSchemaFromCreateCollection_Call {
+	_c.Call.Return(_a0)
+	return _c
+}
+
+func (_c *MockShardManager_AppendNewCollectionSchemaFromCreateCollection_Call) RunAndReturn(run func(message.ImmutableCreateCollectionMessageV1) error) *MockShardManager_AppendNewCollectionSchemaFromCreateCollection_Call {
+	_c.Call.Return(run)
+	return _c
+}
+
+// AlterCollection provides a mock function with given fields: msg
+func (_m *MockShardManager) AlterCollection(msg message.MutableAlterCollectionMessageV2) ([]int64, error) {
+	ret := _m.Called(msg)
+
+	if len(ret) == 0 {
+		panic("no return value specified for AlterCollection")
+	}
+
+	var r0 []int64
+	var r1 error
+	if rf, ok := ret.Get(0).(func(message.MutableAlterCollectionMessageV2) ([]int64, error)); ok {
+		return rf(msg)
+	}
+	if rf, ok := ret.Get(0).(func(message.MutableAlterCollectionMessageV2) []int64); ok {
+		r0 = rf(msg)
+	} else {
+		if ret.Get(0) != nil {
+			r0 = ret.Get(0).([]int64)
+		}
+	}
+	if rf, ok := ret.Get(1).(func(message.MutableAlterCollectionMessageV2) error); ok {
+		r1 = rf(msg)
+	} else {
+		r1 = ret.Error(1)
+	}
+
+	return r0, r1
+}
+
+// MockShardManager_AlterCollection_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'AlterCollection'
+type MockShardManager_AlterCollection_Call struct {
+	*mock.Call
+}
+
+// AlterCollection is a helper method to define mock.On call
+//   - msg message.MutableAlterCollectionMessageV2
+func (_e *MockShardManager_Expecter) AlterCollection(msg interface{}) *MockShardManager_AlterCollection_Call {
+	return &MockShardManager_AlterCollection_Call{Call: _e.mock.On("AlterCollection", msg)}
+}
+
+func (_c *MockShardManager_AlterCollection_Call) Run(run func(msg message.MutableAlterCollectionMessageV2)) *MockShardManager_AlterCollection_Call {
+	_c.Call.Run(func(args mock.Arguments) {
+		run(args[0].(message.MutableAlterCollectionMessageV2))
+	})
+	return _c
+}
+
+func (_c *MockShardManager_AlterCollection_Call) Return(_a0 []int64, _a1 error) *MockShardManager_AlterCollection_Call {
+	_c.Call.Return(_a0, _a1)
+	return _c
+}
+
+func (_c *MockShardManager_AlterCollection_Call) RunAndReturn(run func(message.MutableAlterCollectionMessageV2) ([]int64, error)) *MockShardManager_AlterCollection_Call {
+	_c.Call.Return(run)
+	return _c
+}
+
 // ApplyDelete provides a mock function with given fields: msg
 func (_m *MockShardManager) ApplyDelete(msg message.MutableDeleteMessageV1) error {
 	ret := _m.Called(msg)
@@ -301,6 +450,63 @@ func (_c *MockShardManager_CheckIfCollectionExists_Call) RunAndReturn(run func(i
 	return _c
 }

+// CheckIfCollectionSchemaVersionMatch provides a mock function with given fields: collectionID, schemaVersion
+func (_m *MockShardManager) CheckIfCollectionSchemaVersionMatch(collectionID int64, schemaVersion int32) (int32, error) {
+	ret := _m.Called(collectionID, schemaVersion)
+
+	if len(ret) == 0 {
+		panic("no return value specified for CheckIfCollectionSchemaVersionMatch")
+	}
+
+	var r0 int32
+	var r1 error
+	if rf, ok := ret.Get(0).(func(int64, int32) (int32, error)); ok {
+		return rf(collectionID, schemaVersion)
+	}
+	if rf, ok := ret.Get(0).(func(int64, int32) int32); ok {
+		r0 = rf(collectionID, schemaVersion)
+	} else {
+		r0 = ret.Get(0).(int32)
+	}
+
+	if rf, ok := ret.Get(1).(func(int64, int32) error); ok {
+		r1 = rf(collectionID, schemaVersion)
+	} else {
+		r1 = ret.Error(1)
+	}
+
+	return r0, r1
+}
+
+// MockShardManager_CheckIfCollectionSchemaVersionMatch_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'CheckIfCollectionSchemaVersionMatch'
+type MockShardManager_CheckIfCollectionSchemaVersionMatch_Call struct {
+	*mock.Call
+}
+
+// CheckIfCollectionSchemaVersionMatch is a helper method to define mock.On call
+//   - collectionID int64
+//   - schemaVersion int32
+func (_e *MockShardManager_Expecter) CheckIfCollectionSchemaVersionMatch(collectionID interface{}, schemaVersion interface{}) *MockShardManager_CheckIfCollectionSchemaVersionMatch_Call {
+	return &MockShardManager_CheckIfCollectionSchemaVersionMatch_Call{Call: _e.mock.On("CheckIfCollectionSchemaVersionMatch", collectionID, schemaVersion)}
+}
+
+func (_c *MockShardManager_CheckIfCollectionSchemaVersionMatch_Call) Run(run func(collectionID int64, schemaVersion int32)) *MockShardManager_CheckIfCollectionSchemaVersionMatch_Call {
+	_c.Call.Run(func(args mock.Arguments) {
+		run(args[0].(int64), args[1].(int32))
+	})
+	return _c
+}
+
+func (_c *MockShardManager_CheckIfCollectionSchemaVersionMatch_Call) Return(_a0 int32, _a1 error) *MockShardManager_CheckIfCollectionSchemaVersionMatch_Call {
+	_c.Call.Return(_a0, _a1)
+	return _c
+}
+
+func (_c *MockShardManager_CheckIfCollectionSchemaVersionMatch_Call) RunAndReturn(run func(int64, int32) (int32, error)) *MockShardManager_CheckIfCollectionSchemaVersionMatch_Call {
+	_c.Call.Return(run)
+	return _c
+}
+
 // CheckIfPartitionCanBeCreated provides a mock function with given fields: uniquePartitionKey
 func (_m *MockShardManager) CheckIfPartitionCanBeCreated(uniquePartitionKey shards.PartitionUniqueKey) error {
 	ret := _m.Called(uniquePartitionKey)
@@ -37,6 +37,7 @@ type insertTask struct {
 	partitionKeys   *schemapb.FieldData
 	schemaTimestamp uint64
 	collectionID    int64
+	schemaVersion   int32
 }

 // TraceCtx returns insertTask context
@@ -155,6 +156,7 @@ func (it *insertTask) PreExecute(ctx context.Context) error {
 		return merr.WrapErrAsInputErrorWhen(err, merr.ErrCollectionNotFound, merr.ErrDatabaseNotFound)
 	}
 	it.schema = schema.CollectionSchema
+	it.schemaVersion = schema.Version

 	if err := genFunctionFields(ctx, it.insertMsg, schema, false); err != nil {
 		return err
@@ -11,6 +11,7 @@ import (
 	"github.com/milvus-io/milvus-proto/go-api/v2/schemapb"
 	"github.com/milvus-io/milvus/internal/distributed/streaming"
 	"github.com/milvus-io/milvus/internal/util/hookutil"
+	"github.com/milvus-io/milvus/internal/util/streamingutil/status"
 	"github.com/milvus-io/milvus/pkg/v2/log"
 	"github.com/milvus-io/milvus/pkg/v2/mq/msgstream"
 	"github.com/milvus-io/milvus/pkg/v2/streaming/util/message"
@@ -64,9 +65,9 @@ func (it *insertTask) Execute(ctx context.Context) error {
 	// start to repack insert data
 	var msgs []message.MutableMessage
 	if it.partitionKeys == nil {
-		msgs, err = repackInsertDataForStreamingService(it.TraceCtx(), channelNames, it.insertMsg, it.result, ez)
+		msgs, err = repackInsertDataForStreamingService(it.TraceCtx(), channelNames, it.insertMsg, it.result, ez, it.schemaVersion)
 	} else {
-		msgs, err = repackInsertDataWithPartitionKeyForStreamingService(it.TraceCtx(), channelNames, it.insertMsg, it.result, it.partitionKeys, ez)
+		msgs, err = repackInsertDataWithPartitionKeyForStreamingService(it.TraceCtx(), channelNames, it.insertMsg, it.result, it.partitionKeys, ez, it.schemaVersion)
 	}
 	if err != nil {
 		log.Warn("assign segmentID and repack insert data failed", zap.Error(err))
@@ -76,7 +77,11 @@ func (it *insertTask) Execute(ctx context.Context) error {
 	resp := streaming.WAL().AppendMessages(ctx, msgs...)
 	if err := resp.UnwrapFirstError(); err != nil {
 		log.Warn("append messages to wal failed", zap.Error(err))
-		it.result.Status = merr.Status(err)
+		if status.AsStreamingError(err).IsSchemaVersionMismatch() {
+			it.result.Status = merr.Status(merr.ErrCollectionSchemaMismatch)
+		} else {
+			it.result.Status = merr.Status(err)
+		}
 	}
 	// Update result.Timestamp for session consistency.
 	it.result.Timestamp = resp.MaxTimeTick()
@@ -89,6 +94,7 @@ func repackInsertDataForStreamingService(
 	insertMsg *msgstream.InsertMsg,
 	result *milvuspb.MutationResult,
 	ez *message.CipherConfig,
+	schemaVersion int32,
 ) ([]message.MutableMessage, error) {
 	messages := make([]message.MutableMessage, 0)

@@ -118,6 +124,7 @@ func repackInsertDataForStreamingService(
 							BinarySize:  0, // TODO: current not used, message estimate size is used.
 						},
 					},
+					SchemaVersion: schemaVersion,
 				}).
 				WithBody(insertRequest).
 				WithCipher(ez).
@@ -138,6 +145,7 @@ func repackInsertDataWithPartitionKeyForStreamingService(
 	result *milvuspb.MutationResult,
 	partitionKeys *schemapb.FieldData,
 	ez *message.CipherConfig,
+	schemaVersion int32,
 ) ([]message.MutableMessage, error) {
 	messages := make([]message.MutableMessage, 0)

@@ -199,6 +207,7 @@ func repackInsertDataWithPartitionKeyForStreamingService(
 								BinarySize:  0, // TODO: current not used, message estimate size is used.
 							},
 						},
+						SchemaVersion: schemaVersion,
 					}).
 					WithBody(insertRequest).
 					WithCipher(ez).
@@ -71,6 +71,7 @@ type upsertTask struct {
 	// delete task need use the oldIDs
 	oldIDs          *schemapb.IDs
 	schemaTimestamp uint64
+	schemaVersion   int32

 	// write after read, generate write part by queryPreExecute
 	node types.ProxyComponent
@@ -1273,6 +1274,7 @@ func (it *upsertTask) PreExecute(ctx context.Context) error {
 		return err
 	}
 	it.schema = schema
+	it.schemaVersion = schema.Version

 	err = common.CheckNamespace(schema.CollectionSchema, it.req.Namespace)
 	if err != nil {
@@ -9,6 +9,7 @@ import (

 	"github.com/milvus-io/milvus/internal/distributed/streaming"
 	"github.com/milvus-io/milvus/internal/util/hookutil"
+	"github.com/milvus-io/milvus/internal/util/streamingutil/status"
 	"github.com/milvus-io/milvus/pkg/v2/log"
 	"github.com/milvus-io/milvus/pkg/v2/streaming/util/message"
 	"github.com/milvus-io/milvus/pkg/v2/util/merr"
@@ -41,6 +42,9 @@ func (ut *upsertTask) Execute(ctx context.Context) error {
 	resp := streaming.WAL().AppendMessages(ctx, messages...)
 	if err := resp.UnwrapFirstError(); err != nil {
 		log.Warn("append messages to wal failed", zap.Error(err))
+		if status.AsStreamingError(err).IsSchemaVersionMismatch() {
+			return merr.ErrCollectionSchemaMismatch
+		}
 		return err
 	}
 	// Update result.Timestamp for session consistency.
@@ -83,9 +87,9 @@ func (ut *upsertTask) packInsertMessage(ctx context.Context, ez *message.CipherC
 	// start to repack insert data
 	var msgs []message.MutableMessage
 	if ut.partitionKeys == nil {
-		msgs, err = repackInsertDataForStreamingService(ut.TraceCtx(), channelNames, ut.upsertMsg.InsertMsg, ut.result, ez)
+		msgs, err = repackInsertDataForStreamingService(ut.TraceCtx(), channelNames, ut.upsertMsg.InsertMsg, ut.result, ez, ut.schemaVersion)
 	} else {
-		msgs, err = repackInsertDataWithPartitionKeyForStreamingService(ut.TraceCtx(), channelNames, ut.upsertMsg.InsertMsg, ut.result, ut.partitionKeys, ez)
+		msgs, err = repackInsertDataWithPartitionKeyForStreamingService(ut.TraceCtx(), channelNames, ut.upsertMsg.InsertMsg, ut.result, ut.partitionKeys, ez, ut.schemaVersion)
 	}
 	if err != nil {
 		log.Warn("assign segmentID and repack insert data failed", zap.Error(err))
@@ -44,11 +44,12 @@ type msgHandlerImpl struct {
 func (impl *msgHandlerImpl) HandleCreateSegment(ctx context.Context, createSegmentMsg message.ImmutableCreateSegmentMessageV2) error {
 	vchannel := createSegmentMsg.VChannel()
 	h := createSegmentMsg.Header()
+
 	if err := impl.createNewGrowingSegment(ctx, vchannel, h); err != nil {
 		return err
 	}
 	logger := log.With(log.FieldMessage(createSegmentMsg))
-	if err := impl.wbMgr.CreateNewGrowingSegment(ctx, vchannel, h.PartitionId, h.SegmentId); err != nil {
+	if err := impl.wbMgr.CreateNewGrowingSegment(ctx, vchannel, h.PartitionId, h.SegmentId, h.SchemaVersion); err != nil {
 		logger.Warn("fail to create new growing segment")
 		return err
 	}
@@ -83,12 +84,13 @@ func (impl *msgHandlerImpl) createNewGrowingSegment(ctx context.Context, vchanne
 			Vchannel:             vchannel,
 			StorageVersion:       h.StorageVersion,
 			IsCreatedByStreaming: true,
+			SchemaVersion:        h.SchemaVersion,
 		})
 		if err := merr.CheckRPCCall(resp, err); err != nil {
 			logger.Warn("failed to alloc growing segment at datacoord")
 			return errors.Wrap(err, "failed to alloc growing segment at datacoord")
 		}
-		logger.Info("alloc growing segment at datacoord")
+		logger.Info("alloc growing segment at datacoord", zap.Int32("schemaVersion", h.SchemaVersion))
 		return nil
 	}, retry.AttemptAlways())
 }
@@ -16,6 +16,7 @@ import (
 	"go.uber.org/zap"

 	"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/mocks/mock_metastore"
 	"github.com/milvus-io/milvus/internal/streamingnode/server/resource"
 	"github.com/milvus-io/milvus/internal/streamingnode/server/wal"
@@ -172,7 +173,9 @@ func (f *testOneWALFramework) Run() {
 				CollectionId: 100,
 				PartitionIds: []int64{200},
 			}).
-			WithBody(&msgpb.CreateCollectionRequest{}).
+			WithBody(&msgpb.CreateCollectionRequest{
+				CollectionSchema: &schemapb.CollectionSchema{Name: "test_collection_100"},
+			}).
 			WithVChannel(testVChannel).
 			MustBuildMutable()

@@ -283,7 +286,9 @@ func (f *testOneWALFramework) testSendCreateCollection(ctx context.Context, w wa
 			CollectionId: 1,
 			PartitionIds: []int64{2},
 		}).
-		WithBody(&msgpb.CreateCollectionRequest{}).
+		WithBody(&msgpb.CreateCollectionRequest{
+			CollectionSchema: &schemapb.CollectionSchema{Name: "test_collection"},
+		}).
 		WithVChannel(testVChannel).
 		BuildMutable()
 	require.NoError(f.t, err)
@@ -16,7 +16,6 @@ import (
 	"github.com/milvus-io/milvus/pkg/v2/log"
 	"github.com/milvus-io/milvus/pkg/v2/proto/messagespb"
 	"github.com/milvus-io/milvus/pkg/v2/streaming/util/message"
-	"github.com/milvus-io/milvus/pkg/v2/streaming/util/message/messageutil"
 	"github.com/milvus-io/milvus/pkg/v2/util/funcutil"
 )

@@ -146,6 +145,29 @@ func (impl *shardInterceptor) handleInsertMessage(ctx context.Context, msg messa
 	// Assign segment for insert message.
 	// !!! Current implementation a insert message only has one parition, but we need to merge the message for partition-key in future.
 	header := insertMsg.Header()
+	schemaVersion := header.GetSchemaVersion()
+	if correctSchemaVersion, err := impl.shardManager.CheckIfCollectionSchemaVersionMatch(header.GetCollectionId(), schemaVersion); err != nil {
+		if errors.Is(err, shards.ErrCollectionNotFound) {
+			return nil, status.NewUnrecoverableError("collection %d not found", header.GetCollectionId())
+		}
+		if errors.Is(err, shards.ErrCollectionSchemaNotFound) {
+			return nil, status.NewUnrecoverableError("collection %d schema not provided by create collection message", header.GetCollectionId())
+		}
+		if errors.Is(err, shards.ErrCollectionSchemaVersionNotMatch) {
+			impl.shardManager.Logger().Warn("insertMessage schema version mismatch",
+				zap.Int64("collectionID", header.GetCollectionId()),
+				zap.Int32("schemaVersion", schemaVersion),
+				zap.Int32("collectionSchemaVersion", correctSchemaVersion),
+				zap.Error(err))
+			return nil, status.NewSchemaVersionMismatch("schema version mismatch, input schema version: %d, collection schema version: %d",
+				schemaVersion, correctSchemaVersion)
+		}
+		impl.shardManager.Logger().Error("unexpected error from CheckIfCollectionSchemaVersionMatch",
+			zap.Int64("collectionID", header.GetCollectionId()),
+			zap.Int32("schemaVersion", schemaVersion),
+			zap.Error(err))
+		return nil, errors.Wrap(err, "CheckIfCollectionSchemaVersionMatch")
+	}
 	for _, partition := range header.GetPartitions() {
 		if partition.BinarySize == 0 {
 			// binary size should be set at proxy with estimate, but we don't implement it right now.
@@ -241,7 +263,6 @@ func (impl *shardInterceptor) handleSchemaChange(ctx context.Context, msg messag
 	// Modify the header of schema change message, carry with the all flushed segment ids.
 	header.FlushedSegmentIds = segmentIDs
 	schemaChangeMsg.OverwriteHeader(header)
-
 	return appendOp(ctx, msg)
 }

@@ -249,17 +270,20 @@ func (impl *shardInterceptor) handleSchemaChange(ctx context.Context, msg messag
 func (impl *shardInterceptor) handleAlterCollection(ctx context.Context, msg message.MutableMessage, appendOp interceptors.Append) (message.MessageID, error) {
 	putCollectionMsg := message.MustAsMutableAlterCollectionMessageV2(msg)
 	header := putCollectionMsg.Header()
-	var segmentIDs []int64
-	var err error
-	if messageutil.IsSchemaChange(header) {
-		segmentIDs, err = impl.shardManager.FlushAndFenceSegmentAllocUntil(header.GetCollectionId(), msg.TimeTick())
-		if err != nil {
-			return nil, status.NewUnrecoverableError(err.Error())
-		}
+
+	// AlterCollection atomically flushes+fences segments (if schema change) and updates
+	// in-memory schema — all within one critical region of the shard manager.
+	segmentIDs, err := impl.shardManager.AlterCollection(putCollectionMsg)
+	if err != nil {
+		return nil, status.NewUnrecoverableError(err.Error())
+	}
+
+	// Embed flushed segment IDs into the WAL message header before appending.
+	if len(segmentIDs) > 0 {
+		header.FlushedSegmentIds = segmentIDs
+		putCollectionMsg.OverwriteHeader(header)
 	}

-	header.FlushedSegmentIds = segmentIDs
-	putCollectionMsg.OverwriteHeader(header)
 	return appendOp(ctx, msg)
 }

@@ -202,17 +202,48 @@ func TestShardInterceptor(t *testing.T) {
 		WithBody(&msgpb.InsertRequest{}).
 		MustBuildMutable().WithTimeTick(1)

+	shardManager.EXPECT().CheckIfCollectionSchemaVersionMatch(int64(1), int32(0)).Return(int32(0), nil)
 	shardManager.EXPECT().AssignSegment(mock.Anything).Return(&shards.AssignSegmentResult{SegmentID: 1, Acknowledge: atomic.NewInt32(1)}, nil)
 	msgID, err = i.DoAppend(ctx, msg, appender)
 	assert.NoError(t, err)
 	assert.NotNil(t, msgID)

 	shardManager.EXPECT().AssignSegment(mock.Anything).Unset()
+	shardManager.EXPECT().CheckIfCollectionSchemaVersionMatch(int64(1), int32(0)).Return(int32(0), nil)
 	shardManager.EXPECT().AssignSegment(mock.Anything).Return(nil, mockErr)
 	msgID, err = i.DoAppend(ctx, msg, appender)
 	assert.Error(t, err)
 	assert.Nil(t, msgID)

+	// ErrCollectionNotFound from schema version check must surface as an unrecoverable insert error.
+	shardManager.EXPECT().CheckIfCollectionSchemaVersionMatch(int64(1), int32(0)).Unset()
+	shardManager.EXPECT().CheckIfCollectionSchemaVersionMatch(int64(1), int32(0)).Return(int32(-1), shards.ErrCollectionNotFound)
+	msgID, err = i.DoAppend(ctx, msg, appender)
+	assert.Error(t, err)
+	assert.Nil(t, msgID)
+
+	// ErrCollectionSchemaNotFound must also become an unrecoverable insert error.
+	shardManager.EXPECT().CheckIfCollectionSchemaVersionMatch(int64(1), int32(0)).Unset()
+	shardManager.EXPECT().CheckIfCollectionSchemaVersionMatch(int64(1), int32(0)).Return(int32(-1), shards.ErrCollectionSchemaNotFound)
+	msgID, err = i.DoAppend(ctx, msg, appender)
+	assert.Error(t, err)
+	assert.Nil(t, msgID)
+
+	// ErrCollectionSchemaVersionNotMatch must surface as a schema-version-mismatch error
+	// so the proxy can refresh its cache and retry.
+	shardManager.EXPECT().CheckIfCollectionSchemaVersionMatch(int64(1), int32(0)).Unset()
+	shardManager.EXPECT().CheckIfCollectionSchemaVersionMatch(int64(1), int32(0)).Return(int32(5), shards.ErrCollectionSchemaVersionNotMatch)
+	msgID, err = i.DoAppend(ctx, msg, appender)
+	assert.Error(t, err)
+	assert.Nil(t, msgID)
+
+	// Unexpected error from the schema version check must be propagated as-is.
+	shardManager.EXPECT().CheckIfCollectionSchemaVersionMatch(int64(1), int32(0)).Unset()
+	shardManager.EXPECT().CheckIfCollectionSchemaVersionMatch(int64(1), int32(0)).Return(int32(-1), mockErr)
+	msgID, err = i.DoAppend(ctx, msg, appender)
+	assert.Error(t, err)
+	assert.Nil(t, msgID)
+
 	msg = message.NewDeleteMessageBuilderV1().
 		WithVChannel(vchannel).
 		WithHeader(&messagespb.DeleteMessageHeader{
@@ -89,6 +89,7 @@ func (m *partitionManager) GetSegmentManager(segmentID int64) *segmentAllocManag
 }

 // AssignSegment assigns a segment for a assign segment request.
+// req.SchemaVersion must be set by the caller before invoking.
 func (m *partitionManager) AssignSegment(req *AssignSegmentRequest) (*AssignSegmentResult, error) {
 	// !!! We have promised that the fencedAssignTimeTick is always less than new incoming insert request by Barrier TimeTick of ManualFlush.
 	// So it's just a promise check here.
@@ -211,6 +212,6 @@ func (m *partitionManager) assignSegment(req *AssignSegmentRequest) (*AssignSegm

 	// There is no segment can be allocated for the insert request.
 	// Ask a new pending segment to insert.
-	m.asyncAllocSegment()
+	m.asyncAllocSegment(req.SchemaVersion)
 	return nil, ErrWaitForNewSegment
 }
@@ -18,7 +18,7 @@ import (
 )

 // asyncAllocSegment allocates a new growing segment asynchronously.
-func (m *partitionManager) asyncAllocSegment() {
+func (m *partitionManager) asyncAllocSegment(schemaVersion int32) {
 	if m.onAllocating != nil {
 		m.Logger().Debug("segment alloc worker is already on allocating")
 		// manager is already on allocating.
@@ -27,11 +27,12 @@ func (m *partitionManager) asyncAllocSegment() {
 	// Create a notifier to notify the waiter when the allocation is done.
 	m.onAllocating = make(chan struct{})
 	w := &segmentAllocWorker{
-		ctx:          m.ctx,
-		collectionID: m.collectionID,
-		partitionID:  m.partitionID,
-		vchannel:     m.vchannel,
-		wal:          m.wal.Get(),
+		ctx:           m.ctx,
+		collectionID:  m.collectionID,
+		partitionID:   m.partitionID,
+		vchannel:      m.vchannel,
+		wal:           m.wal.Get(),
+		schemaVersion: schemaVersion,
 	}
 	w.SetLogger(m.Logger())
 	// It should always done asynchronously.
@@ -52,6 +53,7 @@ type segmentAllocWorker struct {
 	segmentID      uint64            // allocated segment ID
 	storageVersion int64             // storage version determined at first attempt
 	limitation     segmentLimitation // segment limitation determined at first attempt
+	schemaVersion  int32
 }

 // do is the main loop of the segment allocation worker.
@@ -111,6 +113,7 @@ func (w *segmentAllocWorker) doOnce() error {
 			MaxRows:        w.limitation.SegmentRows,
 			MaxSegmentSize: w.limitation.SegmentSize,
 			Level:          datapb.SegmentLevel_L1,
+			SchemaVersion:  w.schemaVersion,
 		}).
 		WithBody(&message.CreateSegmentMessageBody{}).
 		MustBuildMutable()
@@ -16,19 +16,22 @@ import (
 	"github.com/milvus-io/milvus/internal/streamingnode/server/wal/recovery"
 	"github.com/milvus-io/milvus/pkg/v2/common"
 	"github.com/milvus-io/milvus/pkg/v2/log"
+	"github.com/milvus-io/milvus/pkg/v2/proto/streamingpb"
 	"github.com/milvus-io/milvus/pkg/v2/streaming/util/types"
 	"github.com/milvus-io/milvus/pkg/v2/util/syncutil"
 )

 var (
-	ErrCollectionExists   = errors.New("collection exists")
-	ErrCollectionNotFound = errors.New("collection not found")
-	ErrPartitionExists    = errors.New("partition exists")
-	ErrPartitionNotFound  = errors.New("partition not found")
-	ErrSegmentExists      = errors.New("segment exists")
-	ErrSegmentNotFound    = errors.New("segment not found")
-	ErrSegmentOnGrowing   = errors.New("segment on growing")
-	ErrFencedAssign       = errors.New("fenced assign")
+	ErrCollectionExists                = errors.New("collection exists")
+	ErrCollectionNotFound              = errors.New("collection not found")
+	ErrCollectionSchemaNotFound        = errors.New("collection schema not found")
+	ErrCollectionSchemaVersionNotMatch = errors.New("collection schema version not match")
+	ErrPartitionExists                 = errors.New("partition exists")
+	ErrPartitionNotFound               = errors.New("partition not found")
+	ErrSegmentExists                   = errors.New("segment exists")
+	ErrSegmentNotFound                 = errors.New("segment not found")
+	ErrSegmentOnGrowing                = errors.New("segment on growing")
+	ErrFencedAssign                    = errors.New("fenced assign")

 	ErrTimeTickTooOld    = errors.New("time tick is too old")
 	ErrWaitForNewSegment = errors.New("wait for new segment")
@@ -159,9 +162,15 @@ func newCollectionInfos(recoverInfos *recovery.RecoverySnapshot) map[int64]*Coll
 		}
 		// add all partitions id into the collection info.
 		currentPartition[common.AllPartitionsID] = struct{}{}
+		// Only keep the latest schema, as shard_interceptor only needs the current write view
+		var latestSchema *streamingpb.CollectionSchemaOfVChannel
+		if len(vchannelInfo.CollectionInfo.Schemas) > 0 {
+			latestSchema = vchannelInfo.CollectionInfo.Schemas[len(vchannelInfo.CollectionInfo.Schemas)-1]
+		}
 		collectionInfoMap[vchannelInfo.CollectionInfo.CollectionId] = &CollectionInfo{
 			VChannel:     vchannelInfo.Vchannel,
 			PartitionIDs: currentPartition,
+			Schema:       latestSchema,
 		}
 	}
 	return collectionInfoMap
@@ -187,6 +196,20 @@ type shardManagerImpl struct {
 type CollectionInfo struct {
 	VChannel     string
 	PartitionIDs map[int64]struct{}
+	Schema       *streamingpb.CollectionSchemaOfVChannel
+}
+
+// SchemaVersion returns the current collection schema version for the write path.
+// It returns 0 if schema is not set (nil receiver, nil Schema, or nil inner CollectionSchema).
+func (c *CollectionInfo) SchemaVersion() int32 {
+	if c == nil || c.Schema == nil {
+		return 0
+	}
+	s := c.Schema.GetSchema()
+	if s == nil {
+		return 0
+	}
+	return s.GetVersion()
 }

 func (m *shardManagerImpl) Channel() types.PChannelInfo {
@@ -215,6 +238,7 @@ func newCollectionInfo(vchannel string, partitionIDs []int64) *CollectionInfo {
 	info := &CollectionInfo{
 		VChannel:     vchannel,
 		PartitionIDs: make(map[int64]struct{}, len(partitionIDs)),
+		Schema:       nil, // Schema will be set when collection is created or altered
 	}
 	for _, partitionID := range partitionIDs {
 		info.PartitionIDs[partitionID] = struct{}{}
@@ -1,11 +1,14 @@
 package shards

 import (
+	"github.com/cockroachdb/errors"
 	"go.uber.org/zap"

 	"github.com/milvus-io/milvus/internal/streamingnode/server/wal/interceptors/shard/policy"
 	"github.com/milvus-io/milvus/pkg/v2/log"
+	"github.com/milvus-io/milvus/pkg/v2/proto/streamingpb"
 	"github.com/milvus-io/milvus/pkg/v2/streaming/util/message"
+	"github.com/milvus-io/milvus/pkg/v2/streaming/util/message/messageutil"
 )

 // CheckIfCollectionCanBeCreated checks if a collection can be created.
@@ -48,6 +51,7 @@ func (m *shardManagerImpl) CreateCollection(msg message.ImmutableCreateCollectio
 	partitionIDs := msg.Header().PartitionIds
 	vchannel := msg.VChannel()
 	timetick := msg.TimeTick()
+	schema := msg.MustBody().GetCollectionSchema()
 	logger := m.Logger().With(log.FieldMessage(msg))

 	m.mu.Lock()
@@ -58,8 +62,18 @@ func (m *shardManagerImpl) CreateCollection(msg message.ImmutableCreateCollectio
 		return
 	}

-	m.collections[collectionID] = newCollectionInfo(vchannel, partitionIDs)
-	for partitionID := range m.collections[collectionID].PartitionIDs {
+	collectionInfo := newCollectionInfo(vchannel, partitionIDs)
+	// Set schema when creating collection
+	if schema != nil {
+		collectionInfo.Schema = &streamingpb.CollectionSchemaOfVChannel{
+			Schema:             schema,
+			CheckpointTimeTick: timetick,
+			State:              streamingpb.VChannelSchemaState_VCHANNEL_SCHEMA_STATE_NORMAL,
+		}
+	}
+	m.collections[collectionID] = collectionInfo
+
+	for partitionID := range collectionInfo.PartitionIDs {
 		uniqueKey := PartitionUniqueKey{CollectionID: collectionID, PartitionID: partitionID}
 		if _, ok := m.partitionManagers[uniqueKey]; ok {
 			logger.Warn("partition already exists", zap.Int64("partitionID", partitionID))
@@ -119,3 +133,88 @@ func (m *shardManagerImpl) DropCollection(msg message.ImmutableDropCollectionMes
 	logger.Info("collection removed", zap.Int64s("partitionIDs", partitionIDs), zap.Int64s("segmentIDs", segmentIDs))
 	m.updateMetrics()
 }
+
+// AlterCollection handles the alter collection message.
+// It updates the schema if present, all within one critical region before WAL append.
+func (m *shardManagerImpl) AlterCollection(msg message.MutableAlterCollectionMessageV2) ([]int64, error) {
+	header := msg.Header()
+	collectionID := header.CollectionId
+	timetick := msg.TimeTick()
+	logger := m.Logger().With(log.FieldMessage(msg))
+
+	m.mu.Lock()
+	defer m.mu.Unlock()
+
+	if err := m.checkIfCollectionExists(collectionID); err != nil {
+		logger.Warn("collection not found when altering collection", zap.Int64("collectionID", collectionID))
+		return nil, err
+	}
+
+	// For schema changes: flush/fence segment allocation and update in-memory schema
+	// atomically within this critical region. Both operations share the same
+	// IsSchemaChange gate so live path and recovery replay stay consistent.
+	var segmentIDs []int64
+	if messageutil.IsSchemaChange(header) {
+		var err error
+		segmentIDs, err = m.flushAndFenceSegmentAllocUntil(collectionID, timetick)
+		if err != nil {
+			return nil, err
+		}
+		logger.Info("flushed segments on schema change", zap.Int64s("segmentIDs", segmentIDs))
+
+		schema := msg.MustBody().Updates.Schema
+		if schema == nil {
+			// UpdateMask says schema changed but the body carries no schema —
+			// malformed message; fail fast to avoid nil-pointer dereferences
+			// in downstream GetSchema paths.
+			logger.Error("schema change indicated by UpdateMask but schema body is nil",
+				zap.Int64("collectionID", collectionID))
+			return nil, errors.New("schema change message has nil schema body")
+		}
+		collectionInfo := m.collections[collectionID]
+		collectionInfo.Schema = &streamingpb.CollectionSchemaOfVChannel{
+			Schema:             schema,
+			CheckpointTimeTick: timetick,
+			State:              streamingpb.VChannelSchemaState_VCHANNEL_SCHEMA_STATE_NORMAL,
+		}
+		logger.Info("updated collection schema in shard manager",
+			zap.Int64("collectionID", collectionID),
+			zap.Int32("schemaVersion", schema.GetVersion()),
+			zap.Uint64("checkpointTimeTick", timetick))
+	}
+
+	return segmentIDs, nil
+}
+
+func (m *shardManagerImpl) CheckIfCollectionSchemaVersionMatch(collectionID int64, schemaVersion int32) (int32, error) {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+
+	return m.checkIfCollectionSchemaVersionMatch(collectionID, schemaVersion)
+}
+
+func (m *shardManagerImpl) checkIfCollectionSchemaVersionMatch(collectionID int64, schemaVersion int32) (int32, error) {
+	collectionInfo, ok := m.collections[collectionID]
+	if !ok {
+		m.Logger().Warn("collection not found", zap.Int64("collectionID", collectionID))
+		return -1, ErrCollectionNotFound
+	}
+	// Input schemaVersion 0 means the proxy did not set it (old proxy or old SDK).
+	// Skip the schema presence and version checks for backward compatibility during rolling
+	// upgrades, where a legacy collection may still have Schema == nil when an old proxy writes.
+	if schemaVersion == 0 {
+		return collectionInfo.SchemaVersion(), nil
+	}
+	if collectionInfo.Schema == nil || collectionInfo.Schema.GetSchema() == nil {
+		m.Logger().Warn("collection schema not found", zap.Int64("collectionID", collectionID))
+		return -1, ErrCollectionSchemaNotFound
+	}
+	collectionSchemaVersion := collectionInfo.SchemaVersion()
+	if collectionSchemaVersion != schemaVersion {
+		m.Logger().Warn("collection schema version not match", zap.Int64("collectionID", collectionID),
+			zap.Int32("schemaVersion", schemaVersion),
+			zap.Int32("collectionSchemaVersion", collectionSchemaVersion))
+		return collectionSchemaVersion, ErrCollectionSchemaVersionNotMatch
+	}
+	return collectionSchemaVersion, nil
+}
@@ -48,5 +48,12 @@ type ShardManager interface {

 	AsyncFlushSegment(signal utils.SealSegmentSignal)

+	// AlterCollection updates collection state and, for schema changes, flushes and
+	// fences segment allocation atomically within one critical region.
+	// Returns the IDs of flushed segments (non-empty only for schema changes).
+	AlterCollection(msg message.MutableAlterCollectionMessageV2) ([]int64, error)
+
+	CheckIfCollectionSchemaVersionMatch(collectionID int64, schemaVersion int32) (int32, error)
+
 	Close()
 }
@@ -17,6 +17,7 @@ type AssignSegmentRequest struct {
 	ModifiedMetrics stats.ModifiedMetrics
 	TimeTick        uint64
 	TxnSession      TxnSession
+	SchemaVersion   int32
 }

 // AssignSegmentResult is a result of segment allocation.
@@ -124,6 +125,7 @@ func (m *shardManagerImpl) FlushSegment(msg message.ImmutableFlushMessageV2) {
 }

 // AssignSegment assigns a segment for a assign segment request.
+// It uses the latest schema version from collection info.
 func (m *shardManagerImpl) AssignSegment(req *AssignSegmentRequest) (*AssignSegmentResult, error) {
 	m.mu.Lock()
 	defer m.mu.Unlock()
@@ -133,6 +135,14 @@ func (m *shardManagerImpl) AssignSegment(req *AssignSegmentRequest) (*AssignSegm
 	if !ok {
 		return nil, ErrPartitionNotFound
 	}
+
+	// Populate SchemaVersion from the current collection info into req.
+	// Callers construct req without knowing the schema version; this is the
+	// single place that resolves and stamps it before forwarding to partitionManager.
+	if info := m.collections[req.CollectionID]; info != nil {
+		req.SchemaVersion = info.SchemaVersion()
+	}
+
 	result, err := pm.AssignSegment(req)
 	if err == nil {
 		m.metrics.ObserveInsert(req.ModifiedMetrics.Rows, req.ModifiedMetrics.BinarySize)
@@ -6,8 +6,10 @@ import (

 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/mock"
+	"google.golang.org/protobuf/types/known/fieldmaskpb"

 	"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/mocks/streamingnode/server/mock_wal"
 	"github.com/milvus-io/milvus/internal/streamingnode/server/resource"
 	"github.com/milvus-io/milvus/internal/streamingnode/server/wal"
@@ -288,3 +290,360 @@ func TestShardManager(t *testing.T) {
 	assert.Equal(t, segmentIDs[0], int64(1013))
 	m.Close()
 }
+
+func TestShardManagerSchemaVersionCheck(t *testing.T) {
+	paramtable.Init()
+	resource.InitForTest(t)
+	channel := types.PChannelInfo{
+		Name: "test_channel_schema",
+		Term: 1,
+	}
+	w := mock_wal.NewMockWAL(t)
+	w.EXPECT().Available().RunAndReturn(func() <-chan struct{} {
+		return make(chan struct{})
+	}).Maybe()
+	w.EXPECT().Append(mock.Anything, mock.Anything).Return(&types.AppendResult{
+		MessageID: rmq.NewRmqID(1),
+		TimeTick:  1000,
+	}, nil).Maybe()
+	f := syncutil.NewFuture[wal.WAL]()
+	f.Set(w)
+
+	m := RecoverShardManager(&ShardManagerRecoverParam{
+		ChannelInfo: channel,
+		WAL:         f,
+		InitialRecoverSnapshot: &recovery.RecoverySnapshot{
+			VChannels:          map[string]*streamingpb.VChannelMeta{},
+			SegmentAssignments: map[int64]*streamingpb.SegmentAssignmentMeta{},
+			Checkpoint:         &recovery.WALCheckpoint{TimeTick: 100},
+		},
+		TxnManager: &mockedTxnManager{},
+	}).(*shardManagerImpl)
+
+	// Test 1: CheckIfCollectionSchemaVersionMatch on non-existent collection
+	_, err := m.CheckIfCollectionSchemaVersionMatch(999, 1)
+	assert.ErrorIs(t, err, ErrCollectionNotFound)
+
+	// Test 2: Create collection with schema, then check version match
+	schema := &schemapb.CollectionSchema{
+		Name:    "test_schema_collection",
+		Version: 1,
+	}
+	createMsg := message.NewCreateCollectionMessageBuilderV1().
+		WithVChannel("v_schema").
+		WithHeader(&message.CreateCollectionMessageHeader{
+			CollectionId: 100,
+			PartitionIds: []int64{200},
+		}).
+		WithBody(&msgpb.CreateCollectionRequest{
+			CollectionSchema: schema,
+		}).
+		MustBuildMutable().
+		WithTimeTick(200).
+		WithLastConfirmedUseMessageID().
+		IntoImmutableMessage(rmq.NewRmqID(10))
+	m.CreateCollection(message.MustAsImmutableCreateCollectionMessageV1(createMsg))
+
+	// version match should succeed
+	ver, err := m.CheckIfCollectionSchemaVersionMatch(100, 1)
+	assert.NoError(t, err)
+	assert.Equal(t, int32(1), ver)
+
+	// version 0 (old proxy) should skip check and succeed
+	ver, err = m.CheckIfCollectionSchemaVersionMatch(100, 0)
+	assert.NoError(t, err)
+	assert.Equal(t, int32(1), ver)
+
+	// version mismatch should fail
+	ver, err = m.CheckIfCollectionSchemaVersionMatch(100, 2)
+	assert.ErrorIs(t, err, ErrCollectionSchemaVersionNotMatch)
+	assert.Equal(t, int32(1), ver)
+
+	// Test 3: Create collection without schema (legacy), then check version
+	createMsgNoSchema := message.NewCreateCollectionMessageBuilderV1().
+		WithVChannel("v_no_schema").
+		WithHeader(&message.CreateCollectionMessageHeader{
+			CollectionId: 101,
+			PartitionIds: []int64{201},
+		}).
+		WithBody(&msgpb.CreateCollectionRequest{}).
+		MustBuildMutable().
+		WithTimeTick(300).
+		WithLastConfirmedUseMessageID().
+		IntoImmutableMessage(rmq.NewRmqID(11))
+	m.CreateCollection(message.MustAsImmutableCreateCollectionMessageV1(createMsgNoSchema))
+
+	// collection exists but has no schema
+	_, err = m.CheckIfCollectionSchemaVersionMatch(101, 1)
+	assert.ErrorIs(t, err, ErrCollectionSchemaNotFound)
+
+	// backward-compat: old proxy sends schemaVersion=0 against a schema-less collection,
+	// must succeed and return the legacy version 0 instead of ErrCollectionSchemaNotFound.
+	ver, err = m.CheckIfCollectionSchemaVersionMatch(101, 0)
+	assert.NoError(t, err)
+	assert.Equal(t, int32(0), ver)
+
+	// Test 4: AlterCollection updates schema version
+	updatedSchema := &schemapb.CollectionSchema{
+		Name:    "test_schema_collection",
+		Version: 2,
+	}
+	alterMsg := message.MustAsMutableAlterCollectionMessageV2(
+		message.NewAlterCollectionMessageBuilderV2().
+			WithVChannel("v_schema").
+			WithHeader(&message.AlterCollectionMessageHeader{
+				CollectionId: 100,
+				UpdateMask:   &fieldmaskpb.FieldMask{Paths: []string{message.FieldMaskCollectionSchema}},
+			}).
+			WithBody(&message.AlterCollectionMessageBody{
+				Updates: &message.AlterCollectionMessageUpdates{
+					Schema: updatedSchema,
+				},
+			}).
+			MustBuildMutable().
+			WithTimeTick(500),
+	)
+	_, err = m.AlterCollection(alterMsg)
+	assert.NoError(t, err)
+
+	// now version 2 matches, version 1 doesn't
+	ver, err = m.CheckIfCollectionSchemaVersionMatch(100, 2)
+	assert.NoError(t, err)
+	assert.Equal(t, int32(2), ver)
+
+	ver, err = m.CheckIfCollectionSchemaVersionMatch(100, 1)
+	assert.ErrorIs(t, err, ErrCollectionSchemaVersionNotMatch)
+	assert.Equal(t, int32(2), ver)
+
+	// Test 5: AlterCollection on non-existent collection
+	alterMsgBad := message.MustAsMutableAlterCollectionMessageV2(
+		message.NewAlterCollectionMessageBuilderV2().
+			WithVChannel("v_schema").
+			WithHeader(&message.AlterCollectionMessageHeader{
+				CollectionId: 999,
+				UpdateMask:   &fieldmaskpb.FieldMask{Paths: []string{message.FieldMaskCollectionSchema}},
+			}).
+			WithBody(&message.AlterCollectionMessageBody{
+				Updates: &message.AlterCollectionMessageUpdates{
+					Schema: updatedSchema,
+				},
+			}).
+			MustBuildMutable().
+			WithTimeTick(600),
+	)
+	_, err = m.AlterCollection(alterMsgBad)
+	assert.ErrorIs(t, err, ErrCollectionNotFound)
+
+	// Test 6: outer CollectionSchemaOfVChannel set but inner schema nil — treat as no schema (no panic).
+	m.mu.Lock()
+	m.collections[102] = &CollectionInfo{
+		VChannel: "v_corrupt",
+		PartitionIDs: map[int64]struct{}{
+			common.AllPartitionsID: {},
+		},
+		Schema: &streamingpb.CollectionSchemaOfVChannel{
+			Schema:             nil,
+			CheckpointTimeTick: 1,
+			State:              streamingpb.VChannelSchemaState_VCHANNEL_SCHEMA_STATE_NORMAL,
+		},
+	}
+	m.mu.Unlock()
+	_, err = m.CheckIfCollectionSchemaVersionMatch(102, 1)
+	assert.ErrorIs(t, err, ErrCollectionSchemaNotFound)
+
+	// same backward-compat path: schemaVersion=0 must not be rejected by the nil inner schema.
+	ver, err = m.CheckIfCollectionSchemaVersionMatch(102, 0)
+	assert.NoError(t, err)
+	assert.Equal(t, int32(0), ver)
+
+	m.Close()
+}
+
+// newShardManagerWithGrowingSegment builds a minimal shard manager that has
+// collection collID / partition partID / growing segment segID pre-loaded.
+func newShardManagerWithGrowingSegment(t *testing.T, collID, partID, segID int64) *shardManagerImpl {
+	t.Helper()
+	channel := types.PChannelInfo{Name: "test_alter_channel", Term: 1}
+	w := mock_wal.NewMockWAL(t)
+	w.EXPECT().Available().RunAndReturn(func() <-chan struct{} {
+		return make(chan struct{})
+	}).Maybe()
+	w.EXPECT().Append(mock.Anything, mock.Anything).Return(&types.AppendResult{
+		MessageID: rmq.NewRmqID(1),
+		TimeTick:  9999,
+	}, nil).Maybe()
+	f := syncutil.NewFuture[wal.WAL]()
+	f.Set(w)
+
+	return RecoverShardManager(&ShardManagerRecoverParam{
+		ChannelInfo: channel,
+		WAL:         f,
+		InitialRecoverSnapshot: &recovery.RecoverySnapshot{
+			VChannels: map[string]*streamingpb.VChannelMeta{
+				"v_alter": {
+					Vchannel: "v_alter",
+					State:    streamingpb.VChannelState_VCHANNEL_STATE_NORMAL,
+					CollectionInfo: &streamingpb.CollectionInfoOfVChannel{
+						CollectionId: collID,
+						Partitions: []*streamingpb.PartitionInfoOfVChannel{
+							{PartitionId: partID},
+						},
+					},
+				},
+			},
+			SegmentAssignments: map[int64]*streamingpb.SegmentAssignmentMeta{
+				segID: {
+					CollectionId: collID,
+					PartitionId:  partID,
+					SegmentId:    segID,
+					State:        streamingpb.SegmentAssignmentState_SEGMENT_ASSIGNMENT_STATE_GROWING,
+					Stat: &streamingpb.SegmentAssignmentStat{
+						MaxBinarySize:         200,
+						ModifiedBinarySize:    100,
+						CreateSegmentTimeTick: 50,
+					},
+				},
+			},
+			Checkpoint: &recovery.WALCheckpoint{TimeTick: 100},
+		},
+		TxnManager: &mockedTxnManager{},
+	}).(*shardManagerImpl)
+}
+
+func TestAlterCollectionSchemaChange(t *testing.T) {
+	paramtable.Init()
+	resource.InitForTest(t)
+
+	const (
+		collID   = int64(10)
+		partID   = int64(20)
+		segID    = int64(3001)
+		vchan    = "v_alter"
+		timeTick = uint64(500)
+	)
+
+	schemaV1 := &schemapb.CollectionSchema{Name: "coll", Version: 1}
+	schemaV2 := &schemapb.CollectionSchema{Name: "coll", Version: 2}
+
+	// helper to build an AlterCollection mutable message
+	buildAlterMsg := func(mask []string, schema *schemapb.CollectionSchema, _ int64) message.MutableAlterCollectionMessageV2 {
+		var updateMask *fieldmaskpb.FieldMask
+		if mask != nil {
+			updateMask = &fieldmaskpb.FieldMask{Paths: mask}
+		}
+		return message.MustAsMutableAlterCollectionMessageV2(
+			message.NewAlterCollectionMessageBuilderV2().
+				WithVChannel(vchan).
+				WithHeader(&message.AlterCollectionMessageHeader{
+					CollectionId: collID,
+					UpdateMask:   updateMask,
+				}).
+				WithBody(&message.AlterCollectionMessageBody{
+					Updates: &message.AlterCollectionMessageUpdates{Schema: schema},
+				}).
+				MustBuildMutable().
+				WithTimeTick(timeTick),
+		)
+	}
+
+	t.Run("schema change flushes and fences growing segments and updates schema", func(t *testing.T) {
+		m := newShardManagerWithGrowingSegment(t, collID, partID, segID)
+		defer m.Close()
+
+		// Set initial schema on the collection.
+		m.mu.Lock()
+		m.collections[collID].Schema = &streamingpb.CollectionSchemaOfVChannel{
+			Schema:             schemaV1,
+			CheckpointTimeTick: 100,
+			State:              streamingpb.VChannelSchemaState_VCHANNEL_SCHEMA_STATE_NORMAL,
+		}
+		m.mu.Unlock()
+
+		// AlterCollection with schema change mask + new schema body.
+		segmentIDs, err := m.AlterCollection(buildAlterMsg([]string{message.FieldMaskCollectionSchema}, schemaV2, 20))
+		assert.NoError(t, err)
+		// The growing segment must have been flushed and fenced.
+		assert.Equal(t, []int64{segID}, segmentIDs)
+		// Schema must be updated to v2.
+		ver, err := m.CheckIfCollectionSchemaVersionMatch(collID, 2)
+		assert.NoError(t, err)
+		assert.Equal(t, int32(2), ver)
+	})
+
+	t.Run("non-schema alter does not flush segments", func(t *testing.T) {
+		m := newShardManagerWithGrowingSegment(t, collID, partID, segID)
+		defer m.Close()
+
+		// AlterCollection with a non-schema mask (e.g. properties only).
+		segmentIDs, err := m.AlterCollection(buildAlterMsg([]string{"properties"}, nil, 21))
+		assert.NoError(t, err)
+		// No segments should be flushed.
+		assert.Empty(t, segmentIDs)
+	})
+
+	t.Run("schema change with nil schema body returns error", func(t *testing.T) {
+		m := newShardManagerWithGrowingSegment(t, collID, partID, segID)
+		defer m.Close()
+
+		// AlterCollection with schema change mask but nil schema in body — malformed message.
+		segmentIDs, err := m.AlterCollection(buildAlterMsg([]string{message.FieldMaskCollectionSchema}, nil, 22))
+		assert.Error(t, err)
+		assert.Nil(t, segmentIDs)
+	})
+
+	t.Run("non-schema-change alter with non-nil schema body does not update schema", func(t *testing.T) {
+		m := newShardManagerWithGrowingSegment(t, collID, partID, segID)
+		defer m.Close()
+
+		// Set initial schema v1.
+		m.mu.Lock()
+		m.collections[collID].Schema = &streamingpb.CollectionSchemaOfVChannel{
+			Schema:             schemaV1,
+			CheckpointTimeTick: 100,
+			State:              streamingpb.VChannelSchemaState_VCHANNEL_SCHEMA_STATE_NORMAL,
+		}
+		m.mu.Unlock()
+
+		// UpdateMask does not include schema field, but body carries a schema — should be ignored.
+		segmentIDs, err := m.AlterCollection(buildAlterMsg([]string{"properties"}, schemaV2, 23))
+		assert.NoError(t, err)
+		assert.Empty(t, segmentIDs)
+		// Schema must remain at v1, not updated to v2.
+		ver, err := m.CheckIfCollectionSchemaVersionMatch(collID, 1)
+		assert.NoError(t, err)
+		assert.Equal(t, int32(1), ver)
+	})
+
+	t.Run("alter on non-existent collection returns ErrCollectionNotFound", func(t *testing.T) {
+		m := newShardManagerWithGrowingSegment(t, collID, partID, segID)
+		defer m.Close()
+
+		badMsg := message.MustAsMutableAlterCollectionMessageV2(
+			message.NewAlterCollectionMessageBuilderV2().
+				WithVChannel(vchan).
+				WithHeader(&message.AlterCollectionMessageHeader{
+					CollectionId: 999,
+					UpdateMask:   &fieldmaskpb.FieldMask{Paths: []string{message.FieldMaskCollectionSchema}},
+				}).
+				WithBody(&message.AlterCollectionMessageBody{
+					Updates: &message.AlterCollectionMessageUpdates{Schema: schemaV2},
+				}).
+				MustBuildMutable().
+				WithTimeTick(timeTick),
+		)
+		_, err := m.AlterCollection(badMsg)
+		assert.ErrorIs(t, err, ErrCollectionNotFound)
+	})
+}
+
+func TestCollectionInfoSchemaVersion(t *testing.T) {
+	assert.Equal(t, int32(0), (*CollectionInfo)(nil).SchemaVersion())
+	ci := &CollectionInfo{}
+	assert.Equal(t, int32(0), ci.SchemaVersion())
+	ci.Schema = &streamingpb.CollectionSchemaOfVChannel{}
+	assert.Equal(t, int32(0), ci.SchemaVersion())
+	ci.Schema = &streamingpb.CollectionSchemaOfVChannel{
+		Schema: &schemapb.CollectionSchema{Name: "x", Version: 3},
+	}
+	assert.Equal(t, int32(3), ci.SchemaVersion())
+}
@@ -57,7 +57,7 @@ func (e *StreamingError) IsSkippedOperation() bool {
 func (e *StreamingError) IsUnrecoverable() bool {
 	return e.Code == streamingpb.StreamingCode_STREAMING_CODE_UNRECOVERABLE ||
 		e.IsReplicateViolation() ||
-		e.IsTxnUnavilable()
+		e.IsTxnUnavilable() || e.IsSchemaVersionMismatch()
 }

 // IsReplicateViolation returns true if the error is caused by replicate violation.
@@ -86,6 +86,11 @@ func (e *StreamingError) IsResourceAcquired() bool {
 	return e.Code == streamingpb.StreamingCode_STREAMING_CODE_RESOURCE_ACQUIRED
 }

+// IsSchemaVersionMismatch returns true if the error is caused by schema version mismatch.
+func (e *StreamingError) IsSchemaVersionMismatch() bool {
+	return e.Code == streamingpb.StreamingCode_STREAMING_CODE_SCHEMA_VERSION_MISMATCH
+}
+
 // IsOnShutdown returns true if the error is caused by on shutdown.
 func (e *StreamingError) IsOnShutdown() bool {
 	return e.Code == streamingpb.StreamingCode_STREAMING_CODE_ON_SHUTDOWN
@@ -141,6 +146,11 @@ func NewInvaildArgument(format string, args ...interface{}) *StreamingError {
 	return New(streamingpb.StreamingCode_STREAMING_CODE_INVAILD_ARGUMENT, format, args...)
 }

+// NewSchemaVersionMismatch creates a new StreamingError with code STREAMING_CODE_SCHEMA_VERSION_MISMATCH.
+func NewSchemaVersionMismatch(format string, args ...interface{}) *StreamingError {
+	return New(streamingpb.StreamingCode_STREAMING_CODE_SCHEMA_VERSION_MISMATCH, format, args...)
+}
+
 // NewTransactionExpired creates a new StreamingError with code STREAMING_CODE_TRANSACTION_EXPIRED.
 func NewTransactionExpired(format string, args ...interface{}) *StreamingError {
 	return New(streamingpb.StreamingCode_STREAMING_CODE_TRANSACTION_EXPIRED, format, args...)