fix: validate collection schema mutations in rootcoord (#51368)

## Problem

Milvus supports online collection-schema mutations — add/drop field,
add/drop function, alter field, enable dynamic field — through several
RootCoord DDL paths. Existing segments are written under the committed
schema, so a mutation that reinterprets an existing field (its data
type, nullability, or structural role) or relabels existing data as a
function output makes already-written binlogs read incorrectly. Before
this change there was no single authoritative check that a proposed
schema is a **safe structural successor** of the committed one;
validation was scattered and incomplete across DDL paths, so an unsafe
mutation could reach the WAL and silently corrupt or misinterpret
existing data.

## What this PR does

Adds an authoritative, validation-only, fail-closed gate
`ValidateSchemaEvolution(oldSchema, newSchema)`, invoked in every
RootCoord schema-mutating callback **before** any side effect (analyzer
reservation, bound-index allocation, WAL broadcast). It rejects:

- in-place field reinterpretation — name, data type, element type,
nullability, or structural role (primary / partition / clustering key,
autoID, dynamic);
- relabeling an existing field as a function output — function outputs
must use brand-new fields, never existing data;
- unsafe field additions — a newly added field may not arrive already
marked primary / partition / clustering key, function output, or
dynamic;
- unsafe drops of protected fields;
- more than one clustering key;
- a dynamic field that is also a function output;
- corruption of the reserved `max_field_id` watermark.

Property-only / TTL updates, analyzer rollback, BM25 / MinHash
`add_function_field`, and legacy function Drop / detach paths are
preserved unchanged.

## Scope

Non-function structural invariants only. Function-graph validation
(single producer, output binding, alter-time immutability, cascade) is
intentionally **out of scope here and owned by #51360**. This PR does
not include schema propagation, Proxy barriers, DML draining, backfill /
readiness, index promotion, TextEmbedding enablement, protobuf,
configuration, or C++ changes.

## Validation

- `go test -tags dynamic,test -gcflags='all=-N -l'
./internal/util/schemautil -count=1` — PASS (the gate's own unit tests)
- `git diff --check origin/master...HEAD` — PASS

issue: #51340

---------

Signed-off-by: xiaofanluan <xf@hjjaq.com>
Co-authored-by: xiaofanluan <xf@hjjaq.com>
Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

internal/rootcoord/ddl_callbacks_alter_collection_add_field.go

@@ -83,6 +83,9 @@ func (c *Core) broadcastAlterCollectionForAddField(ctx context.Context, req *mil
 	if err := typeutil.ValidateTextRequiresStorageV3(schema, Params.CommonCfg.UseLoonFFI.GetAsBool()); err != nil {
 		return merr.WrapErrParameterInvalidMsg("%s", err.Error())
 	}
+	if err := validateSchemaEvolution(coll, schema); err != nil {
+		return err
+	}
 
 	cacheExpirations, err := c.getCacheExpireForCollection(ctx, req.GetDbName(), req.GetCollectionName())
 	if err != nil {

internal/rootcoord/ddl_callbacks_alter_collection_add_field_test.go

@@ -221,6 +221,7 @@ func getFieldSchema(fieldName string) []byte {
 	fieldSchema := &schemapb.FieldSchema{
 		Name:     fieldName,
 		DataType: schemapb.DataType_Int64,
+		Nullable: true,
 	}
 	schemaBytes, _ := proto.Marshal(fieldSchema)
 	return schemaBytes

internal/rootcoord/ddl_callbacks_alter_collection_add_struct_field.go

@@ -78,6 +78,9 @@ func (c *Core) broadcastAlterCollectionForAddStructField(ctx context.Context, re
 	schema.StructArrayFields = append(schema.StructArrayFields, structArrayField)
 	properties := updateMaxFieldIDProperty(coll.Properties, maxAssignedFieldIDFromSchema(schema))
 	schema.Properties = properties
+	if err := validateSchemaEvolution(coll, schema); err != nil {
+		return err
+	}
 
 	cacheExpirations, err := c.getCacheExpireForCollection(ctx, req.GetDbName(), req.GetCollectionName())
 	if err != nil {

internal/rootcoord/ddl_callbacks_alter_collection_field.go

@@ -119,6 +119,9 @@ func (c *Core) broadcastAlterCollectionV2ForAlterCollectionField(ctx context.Con
 			break
 		}
 	}
+	if err := validateSchemaEvolution(coll, schema); err != nil {
+		return err
+	}
 	cacheExpirations, err := c.getCacheExpireForCollection(ctx, req.GetDbName(), req.GetCollectionName())
 	if err != nil {
 		return err

internal/rootcoord/ddl_callbacks_alter_collection_properties.go

@@ -44,6 +44,10 @@ func (c *Core) broadcastAlterCollectionForAlterCollection(ctx context.Context, r
 		return merr.WrapErrParameterInvalidMsg("can not provide properties and deletekeys at the same time")
 	}
 
+	if err := validateReservedCollectionProperties(req.GetProperties(), req.GetDeleteKeys()); err != nil {
+		return err
+	}
+
 	if hookutil.ContainsCipherProperties(req.GetProperties(), req.GetDeleteKeys()) {
 		return merr.WrapErrParameterInvalidMsg("can not alter cipher related properties")
 	}
@@ -212,6 +216,18 @@ func (c *Core) broadcastAlterCollectionForAlterCollection(ctx context.Context, r
 	return nil
 }
 
+func validateReservedCollectionProperties(properties []*commonpb.KeyValuePair, deleteKeys []string) error {
+	for _, property := range properties {
+		if property.GetKey() == common.MaxFieldIDKey {
+			return merr.WrapErrParameterInvalidMsg("cannot alter reserved collection property %s", common.MaxFieldIDKey)
+		}
+	}
+	if funcutil.SliceContain(deleteKeys, common.MaxFieldIDKey) {
+		return merr.WrapErrParameterInvalidMsg("cannot delete reserved collection property %s", common.MaxFieldIDKey)
+	}
+	return nil
+}
+
 func validateNamespaceModeImmutable(properties []*commonpb.KeyValuePair, deleteKeys []string) error {
 	for _, prop := range properties {
 		if prop.GetKey() == common.NamespaceModeKey {
@@ -293,6 +309,9 @@ func (c *Core) broadcastAlterCollectionForAlterDynamicField(ctx context.Context,
 	schema.Fields = append(schema.Fields, fieldSchema)
 	properties := updateMaxFieldIDProperty(coll.Properties, fieldSchema.GetFieldID())
 	schema.Properties = properties
+	if err := validateSchemaEvolution(coll, schema); err != nil {
+		return err
+	}
 
 	channels := make([]string, 0, len(coll.VirtualChannelNames)+1)
 	channels = append(channels, streaming.WAL().ControlChannel())
@@ -349,6 +368,9 @@ func (c *Core) broadcastDisableDynamicField(ctx context.Context, req *milvuspb.A
 	schema.EnableDynamicField = false
 	schema.Properties = properties
 	schema.Version = coll.SchemaVersion + 1
+	if err := validateSchemaEvolution(coll, schema); err != nil {
+		return err
+	}
 
 	channels := make([]string, 0, len(coll.VirtualChannelNames)+1)
 	channels = append(channels, streaming.WAL().ControlChannel())

internal/rootcoord/ddl_callbacks_alter_collection_properties_test.go

@@ -19,6 +19,7 @@ import (
 	"context"
 	"testing"
 
+	"github.com/bytedance/mockey"
 	"github.com/cockroachdb/errors"
 	"github.com/stretchr/testify/mock"
 	"github.com/stretchr/testify/require"
@@ -27,13 +28,16 @@ import (
 	"github.com/milvus-io/milvus-proto/go-api/v3/commonpb"
 	"github.com/milvus-io/milvus-proto/go-api/v3/milvuspb"
 	"github.com/milvus-io/milvus-proto/go-api/v3/schemapb"
+	"github.com/milvus-io/milvus/internal/distributed/streaming"
 	imocks "github.com/milvus-io/milvus/internal/mocks"
 	"github.com/milvus-io/milvus/internal/mocks/streamingcoord/server/mock_balancer"
+	"github.com/milvus-io/milvus/internal/mocks/streamingcoord/server/mock_broadcaster"
 	mockrootcoord "github.com/milvus-io/milvus/internal/rootcoord/mocks"
 	"github.com/milvus-io/milvus/internal/streamingcoord/server/balancer/balance"
 	"github.com/milvus-io/milvus/pkg/v3/common"
 	"github.com/milvus-io/milvus/pkg/v3/proto/messagespb"
 	"github.com/milvus-io/milvus/pkg/v3/streaming/util/message"
+	"github.com/milvus-io/milvus/pkg/v3/streaming/util/types"
 	"github.com/milvus-io/milvus/pkg/v3/util"
 	"github.com/milvus-io/milvus/pkg/v3/util/funcutil"
 	"github.com/milvus-io/milvus/pkg/v3/util/merr"
@@ -215,6 +219,50 @@ func TestDDLCallbacksAlterCollectionProperties(t *testing.T) {
 	require.ErrorIs(t, merr.CheckRPCCall(resp, err), merr.ErrParameterInvalid)
 }
 
+func TestAlterCollectionRejectsReservedMaxFieldIDProperty(t *testing.T) {
+	for _, tc := range []struct {
+		name       string
+		properties []*commonpb.KeyValuePair
+		deleteKeys []string
+	}{
+		{
+			name:       "alter",
+			properties: []*commonpb.KeyValuePair{{Key: common.MaxFieldIDKey, Value: "999"}},
+		},
+		{
+			name:       "delete",
+			deleteKeys: []string{common.MaxFieldIDKey},
+		},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			streaming.SetupNoopWALForTest()
+			defer streaming.SetWALForTest(nil)
+
+			coll := new(DDLCallbacksCollectionFunctionTestSuite).createTestCollection()
+			mockMeta := mockrootcoord.NewIMetaTable(t)
+			mockMeta.EXPECT().GetCollectionByName(mock.Anything, "test_db", "test_collection", typeutil.MaxTimestamp, mock.Anything).Return(coll, nil).Maybe()
+			core := newTestCore(withHealthyCode(), withMeta(mockMeta), withBroker(newValidMockBroker()))
+
+			mockBroadcaster := mock_broadcaster.NewMockBroadcastAPI(t)
+			mockBroadcaster.EXPECT().Close().Maybe()
+			mockBroadcaster.EXPECT().Broadcast(mock.Anything, mock.Anything).Return(&types.BroadcastAppendResult{}, nil).Maybe()
+			lockMocker := mockey.Mock((*Core).startBroadcastWithAliasOrCollectionLock).Return(mockBroadcaster, nil).Build()
+			defer lockMocker.UnPatch()
+			cacheMocker := mockey.Mock((*Core).getCacheExpireForCollection).Return(nil, nil).Build()
+			defer cacheMocker.UnPatch()
+
+			err := core.broadcastAlterCollectionForAlterCollection(context.Background(), &milvuspb.AlterCollectionRequest{
+				DbName:         "test_db",
+				CollectionName: "test_collection",
+				Properties:     tc.properties,
+				DeleteKeys:     tc.deleteKeys,
+			})
+			require.ErrorIs(t, err, merr.ErrParameterInvalid)
+			require.ErrorContains(t, err, common.MaxFieldIDKey)
+		})
+	}
+}
+
 func TestValidateNamespaceModeImmutable(t *testing.T) {
 	for _, tc := range []struct {
 		name       string

internal/rootcoord/ddl_callbacks_alter_collection_schema.go

@@ -119,6 +119,9 @@ func (c *Core) broadcastAlterCollectionSchemaAdd(ctx context.Context, broadcaste
 	if err != nil {
 		return err
 	}
+	if err := validateSchemaEvolution(coll, schema); err != nil {
+		return err
+	}
 	if plan.HasFunction() {
 		if err := validator.ValidateFunction(schema, plan.Function.GetName(), true); err != nil {
 			return merr.Wrap(err, "invalid function schema")
@@ -383,6 +386,9 @@ func (c *Core) broadcastAlterCollectionSchemaDrop(ctx context.Context, broadcast
 	if err != nil {
 		return err
 	}
+	if err := validateSchemaEvolution(coll, schema); err != nil {
+		return err
+	}
 
 	cacheExpirations, err := c.getCacheExpireForCollection(ctx, req.GetDbName(), req.GetCollectionName())
 	if err != nil {

internal/rootcoord/ddl_callbacks_alter_collection_schema_test.go

@@ -581,6 +581,7 @@ func TestDDLCallbacksBroadcastAlterCollectionSchema(t *testing.T) {
 	existingOutputFieldSchema := &schemapb.FieldSchema{
 		Name:     "existing_minhash_output",
 		DataType: schemapb.DataType_BinaryVector,
+		Nullable: true,
 		TypeParams: []*commonpb.KeyValuePair{
 			{Key: common.DimKey, Value: "4096"},
 		},
@@ -609,7 +610,7 @@ func TestDDLCallbacksBroadcastAlterCollectionSchema(t *testing.T) {
 	require.ErrorIs(t, merr.CheckRPCCall(resp.GetAlterStatus(), err), merr.ErrParameterInvalid)
 	require.Contains(t, resp.GetAlterStatus().GetReason(), "output field missing_minhash_output_late")
 
-	// happy path: add only a function and mark an existing output field.
+	// Function-only add cannot relabel an existing field as a function output.
 	functionOnlyReq := buildAlterSchemaAddFunctionReq(dbName, collectionName, &schemapb.FunctionSchema{
 		Name:             "minhash_existing_fn",
 		Type:             schemapb.FunctionType_MinHash,
@@ -623,16 +624,18 @@ func TestDDLCallbacksBroadcastAlterCollectionSchema(t *testing.T) {
 		},
 	})
 	resp, err = core.AlterCollectionSchema(ctx, functionOnlyReq)
-	require.NoError(t, merr.CheckRPCCall(resp.GetAlterStatus(), err))
-	assertSchemaVersion(t, ctx, core, dbName, collectionName, 7)
+	alterErr = merr.CheckRPCCall(resp.GetAlterStatus(), err)
+	require.ErrorIs(t, alterErr, merr.ErrParameterInvalid)
+	require.ErrorContains(t, alterErr, "cannot repurpose existing field")
+	assertSchemaVersion(t, ctx, core, dbName, collectionName, 6)
 	coll, err = core.meta.GetCollectionByName(ctx, dbName, collectionName, typeutil.MaxTimestamp, false)
 	require.NoError(t, err)
-	require.Len(t, coll.Functions, 2)
+	require.Len(t, coll.Functions, 1)
 	existingOutputFound := false
 	for _, field := range coll.Fields {
 		if field.Name == "existing_minhash_output" {
 			existingOutputFound = true
-			require.True(t, field.IsFunctionOutput)
+			require.False(t, field.IsFunctionOutput)
 		}
 	}
 	require.True(t, existingOutputFound)
@@ -656,19 +659,19 @@ func TestDDLCallbacksBroadcastAlterCollectionSchema(t *testing.T) {
 	firstAlterReq := buildAlterSchemaReq(dbName, collectionName, "text_input", "sparse_output", "bm25_fn")
 	resp, err = core.AlterCollectionSchema(ctx, firstAlterReq)
 	require.NoError(t, merr.CheckRPCCall(resp.GetAlterStatus(), err))
-	assertSchemaVersion(t, ctx, core, dbName, collectionName, 8)
+	assertSchemaVersion(t, ctx, core, dbName, collectionName, 7)
 
 	// second happy path with DoPhysicalBackfill=true: the flag is ignored by alter schema.
 	secondAlterReq := buildAlterSchemaReq(dbName, collectionName, "text_input", "sparse_output2", "bm25_fn2")
 	secondAlterReq.GetAction().GetAddRequest().DoPhysicalBackfill = true
 	resp, err = core.AlterCollectionSchema(ctx, secondAlterReq)
 	require.NoError(t, merr.CheckRPCCall(resp.GetAlterStatus(), err))
-	assertSchemaVersion(t, ctx, core, dbName, collectionName, 9)
+	assertSchemaVersion(t, ctx, core, dbName, collectionName, 8)
 	updated, err := core.meta.GetCollectionByName(ctx, dbName, collectionName, typeutil.MaxTimestamp, false)
 	require.NoError(t, err)
 	schema := updated.ToCollectionSchemaPB()
 	require.False(t, schema.GetDoPhysicalBackfill())
-	require.EqualValues(t, 9, schema.GetVersion())
+	require.EqualValues(t, 8, schema.GetVersion())
 
 	// case 9: function already exists (same name "bm25_fn")
 	resp, err = core.AlterCollectionSchema(ctx, &milvuspb.AlterCollectionSchemaRequest{
@@ -794,7 +797,7 @@ func TestDDLCallbacksAlterCollectionSchemaAnalyzerFileResourceRefs(t *testing.T)
 	assertFieldNotExists(t, ctx, core, dbName, collectionName, fieldName)
 }
 
-func TestDDLCallbacksAlterCollectionSchemaValidatesFunctionOnlyFinalSchema(t *testing.T) {
+func TestDDLCallbacksAlterCollectionSchemaRejectsFunctionOnlyMaterializedOutput(t *testing.T) {
 	core := initStreamingSystemAndCore(t)
 
 	ctx := context.Background()
@@ -822,6 +825,7 @@ func TestDDLCallbacksAlterCollectionSchemaValidatesFunctionOnlyFinalSchema(t *te
 	invalidOutputBytes, err := proto.Marshal(&schemapb.FieldSchema{
 		Name:     "invalid_minhash_output",
 		DataType: schemapb.DataType_FloatVector,
+		Nullable: true,
 		TypeParams: []*commonpb.KeyValuePair{
 			{Key: common.DimKey, Value: "128"},
 		},
@@ -849,7 +853,7 @@ func TestDDLCallbacksAlterCollectionSchemaValidatesFunctionOnlyFinalSchema(t *te
 	}))
 	alterErr := merr.CheckRPCCall(resp.GetAlterStatus(), err)
 	require.ErrorIs(t, alterErr, merr.ErrParameterInvalid)
-	require.ErrorContains(t, alterErr, "MinHash function output field must be a BinaryVector field")
+	require.ErrorContains(t, alterErr, "cannot repurpose existing field")
 	assertSchemaVersion(t, ctx, core, dbName, collectionName, 2)
 }
 
@@ -1023,6 +1027,7 @@ func TestDDLCallbacksAlterCollectionSchemaAddSkipsSchemaDropReady(t *testing.T)
 	varcharFieldSchema := &schemapb.FieldSchema{
 		Name:     "text_input",
 		DataType: schemapb.DataType_VarChar,
+		Nullable: true,
 		TypeParams: []*commonpb.KeyValuePair{
 			{Key: common.MaxLengthKey, Value: "256"},
 			{Key: common.EnableAnalyzerKey, Value: "true"},

internal/rootcoord/ddl_callbacks_schema_validation.go

@@ -0,0 +1,31 @@
+// Licensed to the LF AI & Data foundation under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package rootcoord
+
+import (
+	"github.com/milvus-io/milvus-proto/go-api/v3/schemapb"
+	"github.com/milvus-io/milvus/internal/metastore/model"
+	"github.com/milvus-io/milvus/internal/util/schemautil"
+)
+
+func validateSchemaEvolution(oldColl *model.Collection, newSchema *schemapb.CollectionSchema) error {
+	var oldSchema *schemapb.CollectionSchema
+	if oldColl != nil {
+		oldSchema = oldColl.ToCollectionSchemaPB()
+	}
+	return schemautil.ValidateSchemaEvolution(oldSchema, newSchema)
+}

internal/rootcoord/ddl_callbacks_schema_validation_test.go

@@ -0,0 +1,218 @@
+// Licensed to the LF AI & Data foundation under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package rootcoord
+
+import (
+	"context"
+	"testing"
+
+	"github.com/bytedance/mockey"
+	"github.com/stretchr/testify/mock"
+	"github.com/stretchr/testify/require"
+	"google.golang.org/protobuf/proto"
+
+	"github.com/milvus-io/milvus-proto/go-api/v3/commonpb"
+	"github.com/milvus-io/milvus-proto/go-api/v3/milvuspb"
+	"github.com/milvus-io/milvus-proto/go-api/v3/schemapb"
+	"github.com/milvus-io/milvus/internal/metastore/model"
+	"github.com/milvus-io/milvus/internal/mocks"
+	"github.com/milvus-io/milvus/internal/mocks/streamingcoord/server/mock_broadcaster"
+	"github.com/milvus-io/milvus/pkg/v3/common"
+	"github.com/milvus-io/milvus/pkg/v3/proto/querypb"
+	"github.com/milvus-io/milvus/pkg/v3/streaming/util/message"
+	"github.com/milvus-io/milvus/pkg/v3/streaming/util/types"
+	"github.com/milvus-io/milvus/pkg/v3/util/funcutil"
+	"github.com/milvus-io/milvus/pkg/v3/util/merr"
+	"github.com/milvus-io/milvus/pkg/v3/util/typeutil"
+)
+
+func TestDDLCallbacksSchemaEvolutionRejectsUnsafeAddCollectionFieldBeforeSideEffects(t *testing.T) {
+	tests := []struct {
+		name            string
+		field           *schemapb.FieldSchema
+		expectsAnalyzer bool
+	}{
+		{
+			name: "non-nullable field without default",
+			field: &schemapb.FieldSchema{
+				Name:     "required",
+				DataType: schemapb.DataType_Int64,
+			},
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			core := initStreamingSystemAndCore(t)
+			ctx := context.Background()
+			dbName := "testDB" + funcutil.RandomString(10)
+			collectionName := "testCollection" + funcutil.RandomString(10)
+			createCollectionForTest(t, ctx, core, dbName, collectionName)
+
+			analyzerCalls := 0
+			if test.expectsAnalyzer {
+				mixCoord := core.mixCoord.(*mocks.MixCoord)
+				mixCoord.EXPECT().ValidateAnalyzer(mock.Anything, mock.Anything).Run(func(context.Context, *querypb.ValidateAnalyzerRequest) {
+					analyzerCalls++
+				}).Return(&querypb.ValidateAnalyzerResponse{Status: merr.Success()}, nil).Maybe()
+			}
+			broadcasts := 0
+			mockBroadcastAPI := mock_broadcaster.NewMockBroadcastAPI(t)
+			mockBroadcastAPI.EXPECT().Close().Return().Maybe()
+			mockBroadcastAPI.EXPECT().Broadcast(mock.Anything, mock.Anything).Run(func(context.Context, message.BroadcastMutableMessage) {
+				broadcasts++
+			}).Return(&types.BroadcastAppendResult{}, nil).Maybe()
+			lockMocker := mockey.Mock((*Core).startBroadcastWithAliasOrCollectionLock).Return(mockBroadcastAPI, nil).Build()
+			t.Cleanup(func() { lockMocker.UnPatch() })
+
+			schemaBytes, err := proto.Marshal(test.field)
+			require.NoError(t, err)
+			resp, err := core.AddCollectionField(ctx, &milvuspb.AddCollectionFieldRequest{
+				DbName:         dbName,
+				CollectionName: collectionName,
+				Schema:         schemaBytes,
+			})
+			require.ErrorIs(t, merr.CheckRPCCall(resp, err), merr.ErrParameterInvalid)
+			assertSchemaVersion(t, ctx, core, dbName, collectionName, 0)
+			assertFieldNotExists(t, ctx, core, dbName, collectionName, test.field.GetName())
+
+			meta := core.meta.(*MetaTable)
+			require.Empty(t, meta.fileResourceRefCnt, "validation must reject before analyzer resource reservation")
+			require.Zero(t, analyzerCalls, "validation must reject before ValidateAnalyzer")
+			require.Zero(t, broadcasts, "validation must reject before broadcast")
+		})
+	}
+}
+
+func TestDDLCallbacksSchemaEvolutionRejectsInPlaceFieldMutation(t *testing.T) {
+	core := initStreamingSystemAndCore(t)
+	ctx := context.Background()
+	dbName := "testDB" + funcutil.RandomString(10)
+	collectionName := "testCollection" + funcutil.RandomString(10)
+
+	resp, err := core.CreateDatabase(ctx, &milvuspb.CreateDatabaseRequest{DbName: dbName})
+	require.NoError(t, merr.CheckRPCCall(resp, err))
+	schemaBytes, err := proto.Marshal(&schemapb.CollectionSchema{
+		Name: collectionName,
+		Fields: []*schemapb.FieldSchema{
+			{Name: "text", DataType: schemapb.DataType_VarChar, TypeParams: []*commonpb.KeyValuePair{{Key: common.MaxLengthKey, Value: "128"}}},
+		},
+	})
+	require.NoError(t, err)
+	resp, err = core.CreateCollection(ctx, &milvuspb.CreateCollectionRequest{DbName: dbName, CollectionName: collectionName, Schema: schemaBytes})
+	require.NoError(t, merr.CheckRPCCall(resp, err))
+
+	// Resizing max_length (grow or shrink) is allowed; removing the bound
+	// entirely is still rejected, since it would drop the write-time bound.
+	resp, err = core.AlterCollectionField(ctx, &milvuspb.AlterCollectionFieldRequest{
+		DbName:         dbName,
+		CollectionName: collectionName,
+		FieldName:      "text",
+		DeleteKeys:     []string{common.MaxLengthKey},
+	})
+	require.ErrorIs(t, merr.CheckRPCCall(resp, err), merr.ErrParameterInvalid)
+	assertSchemaVersion(t, ctx, core, dbName, collectionName, 0)
+	assertFieldProperties(t, ctx, core, dbName, collectionName, "text", common.MaxLengthKey, "128")
+}
+
+func TestDDLCallbacksSchemaEvolutionRejectsGraphBreakingAlterCollectionSchemaDrop(t *testing.T) {
+	core := initStreamingSystemAndCore(t)
+	ctx := context.Background()
+	dbName := "testDB" + funcutil.RandomString(10)
+	collectionName := "testCollection" + funcutil.RandomString(10)
+	createCollectionForTest(t, ctx, core, dbName, collectionName)
+
+	resp, err := core.AlterCollectionSchema(ctx, buildAlterSchemaAddFieldSchemaReq(dbName, collectionName, &schemapb.FieldSchema{
+		Name:     "text_input",
+		DataType: schemapb.DataType_VarChar,
+		Nullable: true,
+		TypeParams: []*commonpb.KeyValuePair{
+			{Key: common.MaxLengthKey, Value: "128"},
+			{Key: common.EnableAnalyzerKey, Value: "true"},
+		},
+	}, false))
+	require.NoError(t, merr.CheckRPCCall(resp.GetAlterStatus(), err))
+	resp, err = core.AlterCollectionSchema(ctx, buildAlterSchemaReq(dbName, collectionName, "text_input", "sparse_output", "bm25"))
+	require.NoError(t, merr.CheckRPCCall(resp.GetAlterStatus(), err))
+	assertSchemaVersion(t, ctx, core, dbName, collectionName, 2)
+
+	resp, err = core.AlterCollectionSchema(ctx, &milvuspb.AlterCollectionSchemaRequest{
+		DbName:         dbName,
+		CollectionName: collectionName,
+		Action: &milvuspb.AlterCollectionSchemaRequest_Action{
+			Op: &milvuspb.AlterCollectionSchemaRequest_Action_DropRequest{
+				DropRequest: &milvuspb.AlterCollectionSchemaRequest_DropRequest{
+					Identifier: &milvuspb.AlterCollectionSchemaRequest_DropRequest_FieldName{FieldName: "text_input"},
+				},
+			},
+		},
+	})
+	require.ErrorIs(t, merr.CheckRPCCall(resp.GetAlterStatus(), err), merr.ErrParameterInvalid)
+	assertSchemaVersion(t, ctx, core, dbName, collectionName, 2)
+	assertFieldExists(t, ctx, core, dbName, collectionName, "text_input", 101)
+}
+
+func TestDDLCallbacksSchemaEvolutionRejectsUnsafeDynamicEnable(t *testing.T) {
+	core := initStreamingSystemAndCore(t)
+	ctx := context.Background()
+	dbName := "testDB" + funcutil.RandomString(10)
+	collectionName := "testCollection" + funcutil.RandomString(10)
+	createCollectionForTest(t, ctx, core, dbName, collectionName)
+
+	meta := core.meta.(*MetaTable)
+	coll, err := meta.GetCollectionByName(ctx, dbName, collectionName, typeutil.MaxTimestamp, false)
+	require.NoError(t, err)
+	meta.ddLock.Lock()
+	meta.collID2Meta[coll.CollectionID].Fields = append(meta.collID2Meta[coll.CollectionID].Fields, &model.Field{
+		FieldID:   101,
+		Name:      common.MetaFieldName,
+		DataType:  schemapb.DataType_JSON,
+		IsDynamic: true,
+		Nullable:  true,
+	})
+	meta.ddLock.Unlock()
+
+	resp, err := core.AlterCollection(ctx, &milvuspb.AlterCollectionRequest{
+		DbName:         dbName,
+		CollectionName: collectionName,
+		Properties:     []*commonpb.KeyValuePair{{Key: common.EnableDynamicSchemaKey, Value: "true"}},
+	})
+	require.ErrorIs(t, merr.CheckRPCCall(resp, err), merr.ErrParameterInvalid)
+	assertSchemaVersion(t, ctx, core, dbName, collectionName, 0)
+}
+
+func TestDDLCallbacksPropertyOnlyAlterSkipsSchemaEvolutionValidation(t *testing.T) {
+	core := initStreamingSystemAndCore(t)
+	ctx := context.Background()
+	dbName := "testDB" + funcutil.RandomString(10)
+	collectionName := "testCollection" + funcutil.RandomString(10)
+	createCollectionForTest(t, ctx, core, dbName, collectionName)
+
+	meta := core.meta.(*MetaTable)
+	coll, err := meta.GetCollectionByName(ctx, dbName, collectionName, typeutil.MaxTimestamp, false)
+	require.NoError(t, err)
+	meta.ddLock.Lock()
+	meta.collID2Meta[coll.CollectionID].Fields[0].IsFunctionOutput = true
+	meta.ddLock.Unlock()
+
+	resp, err := core.AlterCollection(ctx, &milvuspb.AlterCollectionRequest{
+		DbName:         dbName,
+		CollectionName: collectionName,
+		Properties:     []*commonpb.KeyValuePair{{Key: "property_only", Value: "updated"}},
+	})
+	require.NoError(t, merr.CheckRPCCall(resp, err))
+}

internal/util/schemautil/schema_evolution.go

@@ -0,0 +1,540 @@
+// Licensed to the LF AI & Data foundation under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package schemautil
+
+import (
+	"strconv"
+
+	"google.golang.org/protobuf/proto"
+
+	"github.com/milvus-io/milvus-proto/go-api/v3/commonpb"
+	"github.com/milvus-io/milvus-proto/go-api/v3/schemapb"
+	"github.com/milvus-io/milvus/pkg/v3/common"
+	"github.com/milvus-io/milvus/pkg/v3/util/merr"
+	"github.com/milvus-io/milvus/pkg/v3/util/typeutil"
+)
+
+// ValidateSchemaEvolution validates that a proposed schema can safely replace
+// the committed schema without reinterpreting existing rows or leaving an
+// inconsistent field graph.
+func ValidateSchemaEvolution(oldSchema, newSchema *schemapb.CollectionSchema) error {
+	if oldSchema == nil || newSchema == nil {
+		return merr.WrapErrParameterInvalidMsg("old and new collection schemas must not be nil")
+	}
+
+	oldFields, err := buildEvolutionFieldMaps(oldSchema)
+	if err != nil {
+		return err
+	}
+	newFields, err := buildEvolutionFieldMaps(newSchema)
+	if err != nil {
+		return err
+	}
+
+	if err := validateKeptEvolutionFields(oldFields, newFields); err != nil {
+		return err
+	}
+	if err := validateAddedEvolutionFields(oldFields, newFields, newSchema); err != nil {
+		return err
+	}
+	if err := validateDroppedEvolutionFields(oldFields, newFields, newSchema); err != nil {
+		return err
+	}
+	// Function-graph validation (single producer, output binding, alter-immutability,
+	// cascade) is owned by internal/util/function/validator (see #51360); this gate
+	// intentionally only enforces the non-function structural invariants below.
+	if err := validateEvolutionClusteringKey(newSchema); err != nil {
+		return err
+	}
+	if err := validateEvolutionDynamicGraph(newSchema); err != nil {
+		return err
+	}
+	return validateMaxFieldIDEvolution(oldSchema, newSchema, oldFields, newFields)
+}
+
+type evolutionFieldMaps struct {
+	fields          map[int64]*schemapb.FieldSchema
+	structFields    map[int64]*schemapb.StructArrayFieldSchema
+	structSubFields map[int64]struct{}
+	structParents   map[int64]int64
+	allIDs          map[int64]string
+}
+
+func buildEvolutionFieldMaps(schema *schemapb.CollectionSchema) (*evolutionFieldMaps, error) {
+	result := &evolutionFieldMaps{
+		fields:          make(map[int64]*schemapb.FieldSchema),
+		structFields:    make(map[int64]*schemapb.StructArrayFieldSchema),
+		structSubFields: make(map[int64]struct{}),
+		structParents:   make(map[int64]int64),
+		allIDs:          make(map[int64]string),
+	}
+	addID := func(id int64, kind string) error {
+		if previous, ok := result.allIDs[id]; ok {
+			return merr.WrapErrParameterInvalidMsg("duplicate field id %d is used by both %s and %s", id, previous, kind)
+		}
+		result.allIDs[id] = kind
+		return nil
+	}
+
+	for _, field := range schema.GetFields() {
+		if field == nil {
+			return nil, merr.WrapErrParameterInvalidMsg("collection schema contains a nil field")
+		}
+		if err := addID(field.GetFieldID(), "field"); err != nil {
+			return nil, err
+		}
+		result.fields[field.GetFieldID()] = field
+	}
+	for _, structField := range schema.GetStructArrayFields() {
+		if structField == nil {
+			return nil, merr.WrapErrParameterInvalidMsg("collection schema contains a nil struct field")
+		}
+		if err := addID(structField.GetFieldID(), "struct field"); err != nil {
+			return nil, err
+		}
+		result.structFields[structField.GetFieldID()] = structField
+		for _, field := range structField.GetFields() {
+			if field == nil {
+				return nil, merr.WrapErrParameterInvalidMsg("struct field %q contains a nil sub-field", structField.GetName())
+			}
+			if err := addID(field.GetFieldID(), "struct sub-field"); err != nil {
+				return nil, err
+			}
+			result.fields[field.GetFieldID()] = field
+			result.structSubFields[field.GetFieldID()] = struct{}{}
+			result.structParents[field.GetFieldID()] = structField.GetFieldID()
+		}
+	}
+	return result, nil
+}
+
+func validateKeptEvolutionFields(oldFields, newFields *evolutionFieldMaps) error {
+	for id, oldKind := range oldFields.allIDs {
+		if newKind, kept := newFields.allIDs[id]; kept && oldKind != newKind {
+			return merr.WrapErrParameterInvalidMsg("cannot change field id %d from %s to %s", id, oldKind, newKind)
+		}
+	}
+	for id, oldParent := range oldFields.structParents {
+		if newParent, kept := newFields.structParents[id]; kept && oldParent != newParent {
+			return merr.WrapErrParameterInvalidMsg("cannot move struct sub-field id %d from parent %d to parent %d", id, oldParent, newParent)
+		}
+	}
+	for id, oldField := range oldFields.fields {
+		newField, kept := newFields.fields[id]
+		if !kept {
+			continue
+		}
+		if oldFields.allIDs[id] != newFields.allIDs[id] {
+			return merr.WrapErrParameterInvalidMsg("cannot change the kind of field id %d in place", id)
+		}
+		if err := validateKeptEvolutionField(oldField, newField); err != nil {
+			return err
+		}
+	}
+	for id, oldField := range oldFields.structFields {
+		newField, kept := newFields.structFields[id]
+		if !kept {
+			continue
+		}
+		if oldField.GetName() != newField.GetName() {
+			return merr.WrapErrParameterInvalidMsg("cannot rename struct field id %d from %q to %q", id, oldField.GetName(), newField.GetName())
+		}
+		if oldField.GetNullable() != newField.GetNullable() {
+			return merr.WrapErrParameterInvalidMsg("cannot change the nullability of struct field %q in place", oldField.GetName())
+		}
+		oldChildren := make(map[int64]struct{}, len(oldField.GetFields()))
+		for _, child := range oldField.GetFields() {
+			oldChildren[child.GetFieldID()] = struct{}{}
+		}
+		newChildren := make(map[int64]struct{}, len(newField.GetFields()))
+		for _, child := range newField.GetFields() {
+			newChildren[child.GetFieldID()] = struct{}{}
+		}
+		for childID := range oldChildren {
+			if _, kept := newChildren[childID]; !kept {
+				return merr.WrapErrParameterInvalidMsg("cannot drop sub-field id %d from kept struct field %q", childID, oldField.GetName())
+			}
+		}
+		for childID := range newChildren {
+			if _, existed := oldChildren[childID]; !existed {
+				return merr.WrapErrParameterInvalidMsg("cannot add sub-field id %d to kept struct field %q", childID, oldField.GetName())
+			}
+		}
+		if err := validateNumericBoundsEvolution(oldField.GetName(), oldField.GetTypeParams(), newField.GetTypeParams()); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+func validateKeptEvolutionField(oldField, newField *schemapb.FieldSchema) error {
+	name := oldField.GetName()
+	switch {
+	case name != newField.GetName():
+		return merr.WrapErrParameterInvalidMsg("cannot rename field id %d from %q to %q", oldField.GetFieldID(), name, newField.GetName())
+	case oldField.GetDataType() != newField.GetDataType():
+		return merr.WrapErrParameterInvalidMsg("cannot change the data type of field %q in place", name)
+	case oldField.GetElementType() != newField.GetElementType():
+		return merr.WrapErrParameterInvalidMsg("cannot change the element type of field %q in place", name)
+	case oldField.GetNullable() != newField.GetNullable():
+		return merr.WrapErrParameterInvalidMsg("cannot change the nullability of field %q in place", name)
+	case !oldField.GetIsFunctionOutput() && newField.GetIsFunctionOutput():
+		return merr.WrapErrParameterInvalidMsg("cannot repurpose existing field %q as a function output", name)
+	case oldField.GetIsPrimaryKey() != newField.GetIsPrimaryKey(),
+		oldField.GetIsPartitionKey() != newField.GetIsPartitionKey(),
+		oldField.GetIsClusteringKey() != newField.GetIsClusteringKey(),
+		oldField.GetAutoID() != newField.GetAutoID(),
+		oldField.GetIsDynamic() != newField.GetIsDynamic():
+		return merr.WrapErrParameterInvalidMsg("cannot change the structural role of field %q in place", name)
+	}
+	return validateNumericBoundsEvolution(name, oldField.GetTypeParams(), newField.GetTypeParams())
+}
+
+func validateAddedEvolutionFields(oldFields, newFields *evolutionFieldMaps, newSchema *schemapb.CollectionSchema) error {
+	for id, field := range newFields.fields {
+		if _, existed := oldFields.allIDs[id]; existed {
+			continue
+		}
+		if _, isSubField := newFields.structSubFields[id]; isSubField {
+			parentID := newFields.structParents[id]
+			if _, parentExisted := oldFields.structFields[parentID]; parentExisted {
+				return merr.WrapErrParameterInvalidMsg("cannot add sub-field %q to kept struct field id %d", field.GetName(), parentID)
+			}
+			if field.GetIsPrimaryKey() || field.GetAutoID() || field.GetIsPartitionKey() || field.GetIsClusteringKey() || field.GetIsFunctionOutput() || field.GetIsDynamic() {
+				return merr.WrapErrParameterInvalidMsg("cannot add struct sub-field %q with a protected role", field.GetName())
+			}
+			if !field.GetNullable() {
+				return merr.WrapErrParameterInvalidMsg("cannot add non-nullable sub-field %q in a new struct field", field.GetName())
+			}
+			continue
+		}
+		if err := validateAddedEvolutionField(field, newSchema); err != nil {
+			return err
+		}
+	}
+	for id, structField := range newFields.structFields {
+		if _, existed := oldFields.allIDs[id]; existed {
+			continue
+		}
+		if structField.GetFieldID() < common.StartOfUserFieldID || isReservedEvolutionFieldName(structField.GetName()) {
+			return merr.WrapErrParameterInvalidMsg("cannot add system struct field %q online", structField.GetName())
+		}
+		if !structField.GetNullable() {
+			return merr.WrapErrParameterInvalidMsg("cannot add non-nullable struct field %q online", structField.GetName())
+		}
+		if len(structField.GetFields()) == 0 {
+			return merr.WrapErrParameterInvalidMsg("new struct field %q must contain at least one sub-field", structField.GetName())
+		}
+	}
+	return nil
+}
+
+func validateAddedEvolutionField(field *schemapb.FieldSchema, newSchema *schemapb.CollectionSchema) error {
+	name := field.GetName()
+	dynamicEnabled := newSchema.GetEnableDynamicField()
+	if field.GetIsPrimaryKey() {
+		return merr.WrapErrParameterInvalidMsg("cannot add primary key field %q online", name)
+	}
+	if field.GetAutoID() {
+		return merr.WrapErrParameterInvalidMsg("cannot add auto-ID field %q online", name)
+	}
+	if field.GetIsPartitionKey() {
+		return merr.WrapErrParameterInvalidMsg("cannot add partition key field %q online", name)
+	}
+	if field.GetFieldID() < common.StartOfUserFieldID {
+		return merr.WrapErrParameterInvalidMsg("cannot add system field %q online", name)
+	}
+	if isReservedEvolutionFieldName(name) && (!dynamicEnabled || !field.GetIsDynamic() || name != common.MetaFieldName) {
+		return merr.WrapErrParameterInvalidMsg("cannot add system field %q online", name)
+	}
+	if field.GetIsFunctionOutput() {
+		// A function-output field is only legitimate when a function in the new
+		// schema actually produces it. Ordinary AddField does not run the function
+		// graph validator, so without this check a request could persist an orphan
+		// output field with no owning function (later index creation then fails).
+		if !evolutionFieldHasFunctionProducer(newSchema, field) {
+			return merr.WrapErrParameterInvalidMsg("cannot add field %q marked as a function output with no producing function", name)
+		}
+		return nil
+	}
+	if field.GetIsDynamic() {
+		return nil
+	}
+	if !field.GetNullable() && field.GetDefaultValue() == nil {
+		return merr.WrapErrParameterInvalidMsg("cannot add non-nullable field %q without a default value", name)
+	}
+	return nil
+}
+
+func validateDroppedEvolutionFields(oldFields, newFields *evolutionFieldMaps, newSchema *schemapb.CollectionSchema) error {
+	for id, field := range oldFields.fields {
+		if _, kept := newFields.allIDs[id]; kept {
+			continue
+		}
+		if parentID, isSubField := oldFields.structParents[id]; isSubField {
+			if _, parentKept := newFields.structFields[parentID]; parentKept {
+				return merr.WrapErrParameterInvalidMsg("cannot drop sub-field %q from kept struct field id %d", field.GetName(), parentID)
+			}
+		}
+		if err := validateDroppedEvolutionField(field, newSchema); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+func validateDroppedEvolutionField(field *schemapb.FieldSchema, newSchema *schemapb.CollectionSchema) error {
+	name := field.GetName()
+	switch {
+	case field.GetIsPrimaryKey():
+		return merr.WrapErrParameterInvalidMsg("cannot drop primary key field %q", name)
+	case field.GetIsPartitionKey():
+		return merr.WrapErrParameterInvalidMsg("cannot drop partition key field %q", name)
+	case field.GetIsClusteringKey():
+		return merr.WrapErrParameterInvalidMsg("cannot drop clustering key field %q", name)
+	case field.GetFieldID() < common.StartOfUserFieldID:
+		return merr.WrapErrParameterInvalidMsg("cannot drop system field %q", name)
+	case isReservedEvolutionFieldName(name) && (!field.GetIsDynamic() || newSchema.GetEnableDynamicField()):
+		return merr.WrapErrParameterInvalidMsg("cannot drop system field %q", name)
+	}
+	if function := evolutionFunctionReferencing(newSchema, field.GetFieldID()); function != "" {
+		return merr.WrapErrParameterInvalidMsg("cannot drop field %q while function %q still references it", name, function)
+	}
+	if typeutil.IsVectorType(field.GetDataType()) && !evolutionHasVectorField(newSchema) {
+		return merr.WrapErrParameterInvalidMsg("cannot drop the last vector field %q", name)
+	}
+	return nil
+}
+
+// validateEvolutionClusteringKey allows a clustering-key field to be added
+// online, but enforces that a collection is clustered on at most one field.
+func validateEvolutionClusteringKey(schema *schemapb.CollectionSchema) error {
+	clusteringKey := ""
+	for _, field := range schema.GetFields() {
+		if !field.GetIsClusteringKey() {
+			continue
+		}
+		if clusteringKey != "" {
+			return merr.WrapErrParameterInvalidMsg("collection can only have one clustering key, but got both %q and %q", clusteringKey, field.GetName())
+		}
+		clusteringKey = field.GetName()
+	}
+	return nil
+}
+
+func validateEvolutionDynamicGraph(schema *schemapb.CollectionSchema) error {
+	var dynamicField *schemapb.FieldSchema
+	for _, field := range schema.GetFields() {
+		if !field.GetIsDynamic() {
+			continue
+		}
+		if dynamicField != nil {
+			return merr.WrapErrParameterInvalidMsg("collection schema contains more than one dynamic field")
+		}
+		dynamicField = field
+	}
+	if !schema.GetEnableDynamicField() {
+		if dynamicField != nil {
+			return merr.WrapErrParameterInvalidMsg("dynamic field %q exists while dynamic fields are disabled", dynamicField.GetName())
+		}
+		return nil
+	}
+	if dynamicField == nil {
+		return merr.WrapErrParameterInvalidMsg("dynamic fields are enabled but the dynamic field is missing")
+	}
+	if dynamicField.GetName() != common.MetaFieldName || dynamicField.GetDataType() != schemapb.DataType_JSON {
+		return merr.WrapErrParameterInvalidMsg("dynamic field must be the JSON field %q", common.MetaFieldName)
+	}
+	if dynamicField.GetIsFunctionOutput() {
+		return merr.WrapErrParameterInvalidMsg("dynamic field %q cannot be a function output", dynamicField.GetName())
+	}
+	return nil
+}
+
+// validateNumericBoundsEvolution rejects field-param changes that would
+// reinterpret already-written data. A vector dimension is immutable: reading
+// existing vectors under a new dim yields garbage. max_length / max_capacity
+// are write-time bounds only — existing rows stay readable and new inserts are
+// validated against the new bound — so they may grow or shrink freely; they may
+// not be removed or invalidated (that would drop the bound entirely).
+func validateNumericBoundsEvolution(fieldName string, oldParams, newParams []*commonpb.KeyValuePair) error {
+	for _, key := range []string{common.MaxLengthKey, common.MaxCapacityKey} {
+		_, existed, err := evolutionNumericValue(oldParams, key)
+		if err != nil {
+			return merr.WrapErrParameterInvalidMsg("field %q has an invalid old %s bound", fieldName, key)
+		}
+		if !existed {
+			continue
+		}
+		newValue, kept, err := evolutionNumericValue(newParams, key)
+		if err != nil || !kept {
+			return merr.WrapErrParameterInvalidMsg("cannot remove or invalidate %s of field %q", key, fieldName)
+		}
+		if newValue <= 0 {
+			return merr.WrapErrParameterInvalidMsg("%s of field %q must be a positive integer, got %d", key, fieldName, newValue)
+		}
+	}
+
+	oldDim, existed, err := evolutionNumericValue(oldParams, common.DimKey)
+	if err != nil {
+		return merr.WrapErrParameterInvalidMsg("field %q has an invalid old dimension", fieldName)
+	}
+	if !existed {
+		return nil
+	}
+	newDim, kept, err := evolutionNumericValue(newParams, common.DimKey)
+	if err != nil || !kept || oldDim != newDim {
+		return merr.WrapErrParameterInvalidMsg("cannot change or remove the dimension of field %q", fieldName)
+	}
+	return nil
+}
+
+func evolutionNumericValue(params []*commonpb.KeyValuePair, key string) (int64, bool, error) {
+	for _, param := range params {
+		if param.GetKey() != key {
+			continue
+		}
+		value, err := strconv.ParseInt(param.GetValue(), 10, 64)
+		return value, true, err
+	}
+	return 0, false, nil
+}
+
+func validateMaxFieldIDEvolution(oldSchema, newSchema *schemapb.CollectionSchema, oldFields, newFields *evolutionFieldMaps) error {
+	oldProperty, err := evolutionSchemaIntProperty(oldSchema, common.MaxFieldIDKey)
+	if err != nil {
+		return err
+	}
+	newProperty, err := evolutionSchemaIntProperty(newSchema, common.MaxFieldIDKey)
+	if err != nil {
+		return err
+	}
+	if proto.Equal(oldSchema, newSchema) {
+		return nil
+	}
+
+	oldFloor := maxEvolutionFieldID(oldFields)
+	if oldProperty.valid && oldProperty.value > oldFloor {
+		oldFloor = oldProperty.value
+	}
+	expectedMax := oldFloor
+	if newLiveMax := maxEvolutionFieldID(newFields); newLiveMax > expectedMax {
+		expectedMax = newLiveMax
+	}
+	if !newProperty.present || !newProperty.valid {
+		return merr.WrapErrParameterInvalidMsg("schema evolution must contain a valid %s property", common.MaxFieldIDKey)
+	}
+	if newProperty.value != expectedMax {
+		return merr.WrapErrParameterInvalidMsg("%s must be %d after this evolution, got %d", common.MaxFieldIDKey, expectedMax, newProperty.value)
+	}
+
+	for id := range newFields.allIDs {
+		if _, existed := oldFields.allIDs[id]; !existed && id <= oldFloor {
+			return merr.WrapErrParameterInvalidMsg("new field id %d reuses an already allocated field id", id)
+		}
+	}
+	return nil
+}
+
+type evolutionIntProperty struct {
+	value   int64
+	present bool
+	valid   bool
+}
+
+func evolutionSchemaIntProperty(schema *schemapb.CollectionSchema, key string) (evolutionIntProperty, error) {
+	result := evolutionIntProperty{}
+	for _, property := range schema.GetProperties() {
+		if property.GetKey() != key {
+			continue
+		}
+		if result.present {
+			return evolutionIntProperty{}, merr.WrapErrParameterInvalidMsg("schema contains duplicate property %q", key)
+		}
+		result.present = true
+		parsed, err := strconv.ParseInt(property.GetValue(), 10, 64)
+		if err != nil {
+			continue
+		}
+		result.value = parsed
+		result.valid = true
+	}
+	return result, nil
+}
+
+func maxEvolutionFieldID(fields *evolutionFieldMaps) int64 {
+	maxID := int64(common.StartOfUserFieldID)
+	for id := range fields.allIDs {
+		if id > maxID {
+			maxID = id
+		}
+	}
+	return maxID
+}
+
+func isReservedEvolutionFieldName(name string) bool {
+	switch name {
+	case common.RowIDFieldName, common.TimeStampFieldName, common.MetaFieldName, common.NamespaceFieldName, common.VirtualPKFieldName:
+		return true
+	default:
+		return false
+	}
+}
+
+func evolutionFunctionReferencing(schema *schemapb.CollectionSchema, fieldID int64) string {
+	for _, function := range schema.GetFunctions() {
+		for _, inputID := range function.GetInputFieldIds() {
+			if inputID == fieldID {
+				return function.GetName()
+			}
+		}
+		for _, outputID := range function.GetOutputFieldIds() {
+			if outputID == fieldID {
+				return function.GetName()
+			}
+		}
+	}
+	return ""
+}
+
+// evolutionFieldHasFunctionProducer reports whether some function in the schema
+// produces the given field as an output. Field id and name are matched because
+// they are populated at different stages of the add-function-field flow.
+func evolutionFieldHasFunctionProducer(schema *schemapb.CollectionSchema, field *schemapb.FieldSchema) bool {
+	for _, function := range schema.GetFunctions() {
+		for _, outputID := range function.GetOutputFieldIds() {
+			if outputID == field.GetFieldID() {
+				return true
+			}
+		}
+		for _, outputName := range function.GetOutputFieldNames() {
+			if outputName == field.GetName() {
+				return true
+			}
+		}
+	}
+	return false
+}
+
+func evolutionHasVectorField(schema *schemapb.CollectionSchema) bool {
+	for _, field := range typeutil.GetAllFieldSchemas(schema) {
+		if typeutil.IsVectorType(field.GetDataType()) {
+			return true
+		}
+	}
+	return false
+}

internal/util/schemautil/schema_evolution_test.go

@@ -0,0 +1,680 @@
+// Licensed to the LF AI & Data foundation under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package schemautil
+
+import (
+	"strconv"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+	"google.golang.org/protobuf/proto"
+
+	"github.com/milvus-io/milvus-proto/go-api/v3/commonpb"
+	"github.com/milvus-io/milvus-proto/go-api/v3/schemapb"
+	"github.com/milvus-io/milvus/pkg/v3/common"
+)
+
+func TestValidateSchemaEvolutionAllowed(t *testing.T) {
+	tests := []struct {
+		name   string
+		mutate func(*schemapb.CollectionSchema)
+	}{
+		{name: "unchanged", mutate: func(*schemapb.CollectionSchema) {}},
+		{name: "add nullable field", mutate: func(schema *schemapb.CollectionSchema) {
+			schema.Fields = append(schema.Fields, evolutionField(106, "nullable", schemapb.DataType_Int64, true))
+			setMaxFieldID(schema, 106)
+		}},
+		{name: "add field with default", mutate: func(schema *schemapb.CollectionSchema) {
+			field := evolutionField(106, "with_default", schemapb.DataType_Int64, false)
+			field.DefaultValue = &schemapb.ValueField{Data: &schemapb.ValueField_LongData{LongData: 7}}
+			schema.Fields = append(schema.Fields, field)
+			setMaxFieldID(schema, 106)
+		}},
+		{name: "add nullable struct container", mutate: func(schema *schemapb.CollectionSchema) {
+			child := evolutionField(107, "profile[values]", schemapb.DataType_Array, true)
+			child.ElementType = schemapb.DataType_Int64
+			child.TypeParams = []*commonpb.KeyValuePair{{Key: common.MaxCapacityKey, Value: "32"}}
+			schema.StructArrayFields = append(schema.StructArrayFields, &schemapb.StructArrayFieldSchema{
+				FieldID:  106,
+				Name:     "profile",
+				Nullable: true,
+				TypeParams: []*commonpb.KeyValuePair{
+					{Key: common.MaxCapacityKey, Value: "32"},
+				},
+				Fields: []*schemapb.FieldSchema{child},
+			})
+			setMaxFieldID(schema, 107)
+		}},
+		{name: "drop ordinary field", mutate: func(schema *schemapb.CollectionSchema) {
+			schema.Fields = removeEvolutionField(schema.Fields, 105)
+		}},
+		{name: "grow max length", mutate: func(schema *schemapb.CollectionSchema) {
+			setEvolutionTypeParam(evolutionFieldByID(schema, 104), common.MaxLengthKey, "512")
+		}},
+		{name: "grow max capacity", mutate: func(schema *schemapb.CollectionSchema) {
+			setEvolutionTypeParam(evolutionFieldByID(schema, 105), common.MaxCapacityKey, "128")
+		}},
+		{name: "add function output field with producer", mutate: func(schema *schemapb.CollectionSchema) {
+			output := evolutionField(106, "func_output", schemapb.DataType_SparseFloatVector, false)
+			output.IsFunctionOutput = true
+			schema.Fields = append(schema.Fields, output)
+			schema.Functions = append(schema.Functions, &schemapb.FunctionSchema{
+				Name:             "bm25",
+				Type:             schemapb.FunctionType_BM25,
+				OutputFieldNames: []string{"func_output"},
+				OutputFieldIds:   []int64{106},
+			})
+			setMaxFieldID(schema, 106)
+		}},
+		{name: "enable dynamic field", mutate: func(schema *schemapb.CollectionSchema) {
+			schema.EnableDynamicField = true
+			field := evolutionField(106, common.MetaFieldName, schemapb.DataType_JSON, true)
+			field.IsDynamic = true
+			field.DefaultValue = &schemapb.ValueField{Data: &schemapb.ValueField_BytesData{BytesData: []byte("{}")}}
+			schema.Fields = append(schema.Fields, field)
+			setMaxFieldID(schema, 106)
+		}},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			oldSchema := evolutionBaseSchema()
+			newSchema := proto.Clone(oldSchema).(*schemapb.CollectionSchema)
+			test.mutate(newSchema)
+			require.NoError(t, ValidateSchemaEvolution(oldSchema, newSchema))
+		})
+	}
+
+	t.Run("disable dynamic field", func(t *testing.T) {
+		oldSchema := evolutionBaseSchema()
+		oldSchema.EnableDynamicField = true
+		field := evolutionField(106, common.MetaFieldName, schemapb.DataType_JSON, true)
+		field.IsDynamic = true
+		oldSchema.Fields = append(oldSchema.Fields, field)
+		setMaxFieldID(oldSchema, 106)
+		newSchema := proto.Clone(oldSchema).(*schemapb.CollectionSchema)
+		newSchema.EnableDynamicField = false
+		newSchema.Fields = removeEvolutionField(newSchema.Fields, 106)
+		require.NoError(t, ValidateSchemaEvolution(oldSchema, newSchema))
+	})
+
+	t.Run("drop function and clear former output role", func(t *testing.T) {
+		oldSchema := evolutionSchemaWithFunction()
+		newSchema := proto.Clone(oldSchema).(*schemapb.CollectionSchema)
+		newSchema.Functions = nil
+		evolutionFieldByID(newSchema, 106).IsFunctionOutput = false
+		require.NoError(t, ValidateSchemaEvolution(oldSchema, newSchema))
+	})
+
+	t.Run("drop whole struct container", func(t *testing.T) {
+		oldSchema := evolutionSchemaWithStruct()
+		newSchema := proto.Clone(oldSchema).(*schemapb.CollectionSchema)
+		newSchema.StructArrayFields = nil
+		require.NoError(t, ValidateSchemaEvolution(oldSchema, newSchema))
+	})
+
+	for _, test := range []struct {
+		name     string
+		oldValue string
+	}{
+		{name: "repair missing max field id", oldValue: ""},
+		{name: "repair malformed max field id", oldValue: "not-a-number"},
+		{name: "repair stale low max field id", oldValue: "102"},
+	} {
+		t.Run(test.name, func(t *testing.T) {
+			oldSchema := evolutionBaseSchema()
+			if test.oldValue == "" {
+				removeEvolutionProperty(oldSchema, common.MaxFieldIDKey)
+			} else {
+				setEvolutionProperty(oldSchema, common.MaxFieldIDKey, test.oldValue)
+			}
+			newSchema := proto.Clone(oldSchema).(*schemapb.CollectionSchema)
+			setMaxFieldID(newSchema, 105)
+			require.NoError(t, ValidateSchemaEvolution(oldSchema, newSchema))
+		})
+	}
+
+	t.Run("repair stale low max field id using struct child floor", func(t *testing.T) {
+		oldSchema := evolutionSchemaWithStruct()
+		setMaxFieldID(oldSchema, 102)
+		newSchema := proto.Clone(oldSchema).(*schemapb.CollectionSchema)
+		setMaxFieldID(newSchema, 107)
+		require.NoError(t, ValidateSchemaEvolution(oldSchema, newSchema))
+	})
+
+	t.Run("preserve stale high max field id", func(t *testing.T) {
+		oldSchema := evolutionBaseSchema()
+		setMaxFieldID(oldSchema, 120)
+		newSchema := proto.Clone(oldSchema).(*schemapb.CollectionSchema)
+		newSchema.Fields = removeEvolutionField(newSchema.Fields, 105)
+		require.NoError(t, ValidateSchemaEvolution(oldSchema, newSchema))
+	})
+
+	for _, test := range []struct {
+		name  string
+		value string
+	}{
+		{name: "unchanged malformed max field id", value: "not-a-number"},
+		{name: "unchanged stale low max field id", value: "102"},
+	} {
+		t.Run(test.name, func(t *testing.T) {
+			oldSchema := evolutionBaseSchema()
+			setEvolutionProperty(oldSchema, common.MaxFieldIDKey, test.value)
+			newSchema := proto.Clone(oldSchema).(*schemapb.CollectionSchema)
+			require.NoError(t, ValidateSchemaEvolution(oldSchema, newSchema))
+		})
+	}
+}
+
+func TestValidateSchemaEvolutionRejectsKeptFieldReinterpretation(t *testing.T) {
+	tests := []struct {
+		name   string
+		mutate func(*schemapb.FieldSchema)
+	}{
+		{name: "data type", mutate: func(field *schemapb.FieldSchema) { field.DataType = schemapb.DataType_Int32 }},
+		{name: "element type", mutate: func(field *schemapb.FieldSchema) { field.ElementType = schemapb.DataType_Int32 }},
+		{name: "nullability", mutate: func(field *schemapb.FieldSchema) { field.Nullable = false }},
+		{name: "dimension", mutate: func(field *schemapb.FieldSchema) { setEvolutionTypeParam(field, common.DimKey, "256") }},
+		{name: "function output role", mutate: func(field *schemapb.FieldSchema) { field.IsFunctionOutput = true }},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			oldSchema := evolutionBaseSchema()
+			newSchema := proto.Clone(oldSchema).(*schemapb.CollectionSchema)
+			var field *schemapb.FieldSchema
+			switch test.name {
+			case "element type", "nullability":
+				field = evolutionFieldByID(newSchema, 105)
+			case "dimension":
+				field = evolutionFieldByID(newSchema, 103)
+			default:
+				field = evolutionFieldByID(newSchema, 102)
+			}
+			test.mutate(field)
+			require.Error(t, ValidateSchemaEvolution(oldSchema, newSchema))
+		})
+	}
+}
+
+func TestValidateSchemaEvolutionRejectsNonMonotonicBounds(t *testing.T) {
+	tests := []struct {
+		name   string
+		field  int64
+		key    string
+		value  string
+		remove bool
+	}{
+		{name: "remove max length", field: 104, key: common.MaxLengthKey, remove: true},
+		{name: "remove max capacity", field: 105, key: common.MaxCapacityKey, remove: true},
+		{name: "zero max length", field: 104, key: common.MaxLengthKey, value: "0"},
+		{name: "negative max length", field: 104, key: common.MaxLengthKey, value: "-8"},
+		{name: "zero max capacity", field: 105, key: common.MaxCapacityKey, value: "0"},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			oldSchema := evolutionBaseSchema()
+			newSchema := proto.Clone(oldSchema).(*schemapb.CollectionSchema)
+			field := evolutionFieldByID(newSchema, test.field)
+			if test.remove {
+				removeEvolutionTypeParam(field, test.key)
+			} else {
+				setEvolutionTypeParam(field, test.key, test.value)
+			}
+			require.Error(t, ValidateSchemaEvolution(oldSchema, newSchema))
+		})
+	}
+
+	t.Run("shrink max field id", func(t *testing.T) {
+		oldSchema := evolutionBaseSchema()
+		newSchema := proto.Clone(oldSchema).(*schemapb.CollectionSchema)
+		setMaxFieldID(newSchema, 104)
+		require.Error(t, ValidateSchemaEvolution(oldSchema, newSchema))
+	})
+
+	t.Run("remove max field id", func(t *testing.T) {
+		oldSchema := evolutionBaseSchema()
+		newSchema := proto.Clone(oldSchema).(*schemapb.CollectionSchema)
+		removeEvolutionProperty(newSchema, common.MaxFieldIDKey)
+		require.Error(t, ValidateSchemaEvolution(oldSchema, newSchema))
+	})
+
+	t.Run("alter max field id without allocating a field", func(t *testing.T) {
+		oldSchema := evolutionBaseSchema()
+		newSchema := proto.Clone(oldSchema).(*schemapb.CollectionSchema)
+		setMaxFieldID(newSchema, 110)
+		require.Error(t, ValidateSchemaEvolution(oldSchema, newSchema))
+	})
+}
+
+func TestValidateSchemaEvolutionAllowsResizingFieldBounds(t *testing.T) {
+	// max_length / max_capacity are write-time bounds, not a reinterpretation of
+	// already-written data, so they may grow or shrink freely.
+	tests := []struct {
+		name  string
+		field int64
+		key   string
+		value string
+	}{
+		{name: "shrink max length", field: 104, key: common.MaxLengthKey, value: "64"},
+		{name: "grow max length", field: 104, key: common.MaxLengthKey, value: "1024"},
+		{name: "shrink max capacity", field: 105, key: common.MaxCapacityKey, value: "16"},
+		{name: "grow max capacity", field: 105, key: common.MaxCapacityKey, value: "256"},
+	}
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			oldSchema := evolutionBaseSchema()
+			newSchema := proto.Clone(oldSchema).(*schemapb.CollectionSchema)
+			setEvolutionTypeParam(evolutionFieldByID(newSchema, test.field), test.key, test.value)
+			require.NoError(t, ValidateSchemaEvolution(oldSchema, newSchema))
+		})
+	}
+}
+
+func TestValidateSchemaEvolutionRejectsUnsafeAddedFields(t *testing.T) {
+	tests := []struct {
+		name  string
+		field *schemapb.FieldSchema
+	}{
+		{name: "non-nullable field without default", field: evolutionField(106, "required", schemapb.DataType_Int64, false)},
+		{name: "orphan function output", field: func() *schemapb.FieldSchema {
+			field := evolutionField(106, "orphan_output", schemapb.DataType_SparseFloatVector, false)
+			field.IsFunctionOutput = true
+			return field
+		}()},
+		{name: "primary key", field: func() *schemapb.FieldSchema {
+			field := evolutionField(106, "new_pk", schemapb.DataType_Int64, true)
+			field.IsPrimaryKey = true
+			return field
+		}()},
+		{name: "auto id", field: func() *schemapb.FieldSchema {
+			field := evolutionField(106, "auto_id", schemapb.DataType_Int64, true)
+			field.AutoID = true
+			return field
+		}()},
+		{name: "partition key", field: func() *schemapb.FieldSchema {
+			field := evolutionField(106, "partition_key", schemapb.DataType_Int64, true)
+			field.IsPartitionKey = true
+			return field
+		}()},
+		{name: "system field id", field: evolutionField(int64(common.RowIDField), "fake_system", schemapb.DataType_Int64, true)},
+		{name: "system field name", field: evolutionField(106, common.TimeStampFieldName, schemapb.DataType_Int64, true)},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			oldSchema := evolutionBaseSchema()
+			newSchema := proto.Clone(oldSchema).(*schemapb.CollectionSchema)
+			newSchema.Fields = append(newSchema.Fields, test.field)
+			setMaxFieldID(newSchema, 106)
+			require.Error(t, ValidateSchemaEvolution(oldSchema, newSchema))
+		})
+	}
+
+	t.Run("reused dropped field id", func(t *testing.T) {
+		oldSchema := evolutionBaseSchema()
+		newSchema := proto.Clone(oldSchema).(*schemapb.CollectionSchema)
+		newSchema.Fields = removeEvolutionField(newSchema.Fields, 105)
+		newSchema.Fields = append(newSchema.Fields, evolutionField(105, "replacement", schemapb.DataType_Int64, true))
+		require.Error(t, ValidateSchemaEvolution(oldSchema, newSchema))
+	})
+
+	t.Run("legacy schema must establish max field id before adding", func(t *testing.T) {
+		oldSchema := evolutionBaseSchema()
+		removeEvolutionProperty(oldSchema, common.MaxFieldIDKey)
+		newSchema := proto.Clone(oldSchema).(*schemapb.CollectionSchema)
+		newSchema.Fields = append(newSchema.Fields, evolutionField(106, "nullable", schemapb.DataType_Int64, true))
+		require.Error(t, ValidateSchemaEvolution(oldSchema, newSchema))
+	})
+}
+
+func TestValidateSchemaEvolutionRejectsStructSubFieldReparenting(t *testing.T) {
+	oldSchema := evolutionBaseSchema()
+	oldSchema.StructArrayFields = []*schemapb.StructArrayFieldSchema{
+		{
+			FieldID:  106,
+			Name:     "left",
+			Nullable: true,
+			Fields: []*schemapb.FieldSchema{
+				evolutionField(107, "left[value]", schemapb.DataType_Int64, true),
+			},
+		},
+		{FieldID: 108, Name: "right", Nullable: true},
+	}
+	setMaxFieldID(oldSchema, 108)
+	newSchema := proto.Clone(oldSchema).(*schemapb.CollectionSchema)
+	moved := newSchema.StructArrayFields[0].Fields[0]
+	newSchema.StructArrayFields[0].Fields = nil
+	newSchema.StructArrayFields[1].Fields = []*schemapb.FieldSchema{moved}
+
+	require.Error(t, ValidateSchemaEvolution(oldSchema, newSchema))
+}
+
+func TestValidateSchemaEvolutionRejectsIndependentStructSubFieldMutation(t *testing.T) {
+	t.Run("add child to kept container", func(t *testing.T) {
+		oldSchema := evolutionSchemaWithStruct()
+		newSchema := proto.Clone(oldSchema).(*schemapb.CollectionSchema)
+		child := evolutionField(108, "profile[extra]", schemapb.DataType_Array, true)
+		child.ElementType = schemapb.DataType_Int64
+		child.TypeParams = []*commonpb.KeyValuePair{{Key: common.MaxCapacityKey, Value: "16"}}
+		newSchema.StructArrayFields[0].Fields = append(newSchema.StructArrayFields[0].Fields, child)
+		setMaxFieldID(newSchema, 108)
+		require.ErrorContains(t, ValidateSchemaEvolution(oldSchema, newSchema), "cannot add sub-field")
+	})
+
+	t.Run("drop child from kept container", func(t *testing.T) {
+		oldSchema := evolutionSchemaWithStruct()
+		newSchema := proto.Clone(oldSchema).(*schemapb.CollectionSchema)
+		newSchema.StructArrayFields[0].Fields = nil
+		require.ErrorContains(t, ValidateSchemaEvolution(oldSchema, newSchema), "cannot drop sub-field")
+	})
+
+	t.Run("add whole container with non-nullable child", func(t *testing.T) {
+		oldSchema := evolutionBaseSchema()
+		newSchema := proto.Clone(oldSchema).(*schemapb.CollectionSchema)
+		child := evolutionField(107, "profile[values]", schemapb.DataType_Array, false)
+		child.ElementType = schemapb.DataType_Int64
+		child.TypeParams = []*commonpb.KeyValuePair{{Key: common.MaxCapacityKey, Value: "16"}}
+		newSchema.StructArrayFields = []*schemapb.StructArrayFieldSchema{
+			{FieldID: 106, Name: "profile", Nullable: true, Fields: []*schemapb.FieldSchema{child}},
+		}
+		setMaxFieldID(newSchema, 107)
+		require.ErrorContains(t, ValidateSchemaEvolution(oldSchema, newSchema), "non-nullable sub-field")
+	})
+
+	t.Run("add empty whole container", func(t *testing.T) {
+		oldSchema := evolutionBaseSchema()
+		newSchema := proto.Clone(oldSchema).(*schemapb.CollectionSchema)
+		newSchema.StructArrayFields = []*schemapb.StructArrayFieldSchema{
+			{FieldID: 106, Name: "profile", Nullable: true},
+		}
+		setMaxFieldID(newSchema, 106)
+		require.ErrorContains(t, ValidateSchemaEvolution(oldSchema, newSchema), "must contain at least one sub-field")
+	})
+}
+
+func TestValidateSchemaEvolutionRejectsProtectedDrops(t *testing.T) {
+	tests := []struct {
+		name   string
+		mutate func(*schemapb.CollectionSchema)
+	}{
+		{name: "primary key", mutate: func(schema *schemapb.CollectionSchema) { schema.Fields = removeEvolutionField(schema.Fields, 100) }},
+		{name: "system field", mutate: func(schema *schemapb.CollectionSchema) {
+			schema.Fields = removeEvolutionField(schema.Fields, int64(common.RowIDField))
+		}},
+		{name: "last vector field", mutate: func(schema *schemapb.CollectionSchema) { schema.Fields = removeEvolutionField(schema.Fields, 103) }},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			oldSchema := evolutionBaseSchema()
+			newSchema := proto.Clone(oldSchema).(*schemapb.CollectionSchema)
+			test.mutate(newSchema)
+			require.Error(t, ValidateSchemaEvolution(oldSchema, newSchema))
+		})
+	}
+
+	for _, test := range []struct {
+		name string
+		mark func(*schemapb.FieldSchema)
+	}{
+		{name: "partition key", mark: func(field *schemapb.FieldSchema) { field.IsPartitionKey = true }},
+		{name: "clustering key", mark: func(field *schemapb.FieldSchema) { field.IsClusteringKey = true }},
+	} {
+		t.Run(test.name, func(t *testing.T) {
+			oldSchema := evolutionBaseSchema()
+			test.mark(evolutionFieldByID(oldSchema, 104))
+			newSchema := proto.Clone(oldSchema).(*schemapb.CollectionSchema)
+			newSchema.Fields = removeEvolutionField(newSchema.Fields, 104)
+			require.Error(t, ValidateSchemaEvolution(oldSchema, newSchema))
+		})
+	}
+
+	t.Run("surviving function input", func(t *testing.T) {
+		oldSchema := evolutionSchemaWithFunction()
+		newSchema := proto.Clone(oldSchema).(*schemapb.CollectionSchema)
+		newSchema.Fields = removeEvolutionField(newSchema.Fields, 102)
+		require.Error(t, ValidateSchemaEvolution(oldSchema, newSchema))
+	})
+
+	t.Run("surviving function output", func(t *testing.T) {
+		oldSchema := evolutionSchemaWithFunction()
+		newSchema := proto.Clone(oldSchema).(*schemapb.CollectionSchema)
+		newSchema.Fields = removeEvolutionField(newSchema.Fields, 106)
+		require.Error(t, ValidateSchemaEvolution(oldSchema, newSchema))
+	})
+}
+
+func TestValidateSchemaEvolutionRejectsMalformedDynamicGraphs(t *testing.T) {
+	tests := []struct {
+		name   string
+		mutate func(*schemapb.CollectionSchema)
+	}{
+		{name: "enabled without dynamic field", mutate: func(schema *schemapb.CollectionSchema) { schema.EnableDynamicField = true }},
+		{name: "disabled with dynamic field", mutate: func(schema *schemapb.CollectionSchema) {
+			field := evolutionField(106, common.MetaFieldName, schemapb.DataType_JSON, true)
+			field.IsDynamic = true
+			schema.Fields = append(schema.Fields, field)
+			setMaxFieldID(schema, 106)
+		}},
+		{name: "multiple dynamic fields", mutate: func(schema *schemapb.CollectionSchema) {
+			schema.EnableDynamicField = true
+			for id := int64(106); id <= 107; id++ {
+				field := evolutionField(id, common.MetaFieldName+strconv.FormatInt(id, 10), schemapb.DataType_JSON, true)
+				field.IsDynamic = true
+				schema.Fields = append(schema.Fields, field)
+			}
+			setMaxFieldID(schema, 107)
+		}},
+		{name: "dynamic field has wrong type", mutate: func(schema *schemapb.CollectionSchema) {
+			schema.EnableDynamicField = true
+			field := evolutionField(106, common.MetaFieldName, schemapb.DataType_Int64, true)
+			field.IsDynamic = true
+			schema.Fields = append(schema.Fields, field)
+			setMaxFieldID(schema, 106)
+		}},
+		{name: "dynamic field has wrong name", mutate: func(schema *schemapb.CollectionSchema) {
+			schema.EnableDynamicField = true
+			field := evolutionField(106, "dynamic", schemapb.DataType_JSON, true)
+			field.IsDynamic = true
+			schema.Fields = append(schema.Fields, field)
+			setMaxFieldID(schema, 106)
+		}},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			oldSchema := evolutionBaseSchema()
+			newSchema := proto.Clone(oldSchema).(*schemapb.CollectionSchema)
+			test.mutate(newSchema)
+			require.Error(t, ValidateSchemaEvolution(oldSchema, newSchema))
+		})
+	}
+}
+
+func TestValidateSchemaEvolutionClusteringKey(t *testing.T) {
+	longDefault := func() *schemapb.ValueField {
+		return &schemapb.ValueField{Data: &schemapb.ValueField_LongData{LongData: 0}}
+	}
+
+	t.Run("add clustering key allowed", func(t *testing.T) {
+		oldSchema := evolutionBaseSchema()
+		newSchema := proto.Clone(oldSchema).(*schemapb.CollectionSchema)
+		ck := evolutionField(106, "ck", schemapb.DataType_Int64, false)
+		ck.IsClusteringKey = true
+		ck.DefaultValue = longDefault()
+		newSchema.Fields = append(newSchema.Fields, ck)
+		setMaxFieldID(newSchema, 106)
+		require.NoError(t, ValidateSchemaEvolution(oldSchema, newSchema))
+	})
+
+	t.Run("second clustering key rejected", func(t *testing.T) {
+		oldSchema := evolutionBaseSchema()
+		evolutionFieldByID(oldSchema, 104).IsClusteringKey = true
+		newSchema := proto.Clone(oldSchema).(*schemapb.CollectionSchema)
+		ck := evolutionField(106, "ck2", schemapb.DataType_Int64, false)
+		ck.IsClusteringKey = true
+		ck.DefaultValue = longDefault()
+		newSchema.Fields = append(newSchema.Fields, ck)
+		setMaxFieldID(newSchema, 106)
+		require.ErrorContains(t, ValidateSchemaEvolution(oldSchema, newSchema), "one clustering key")
+	})
+}
+
+func evolutionBaseSchema() *schemapb.CollectionSchema {
+	rowID := evolutionField(int64(common.RowIDField), common.RowIDFieldName, schemapb.DataType_Int64, false)
+	rowID.AutoID = true
+	timestamp := evolutionField(int64(common.TimeStampField), common.TimeStampFieldName, schemapb.DataType_Int64, false)
+	pk := evolutionField(100, "pk", schemapb.DataType_Int64, false)
+	pk.IsPrimaryKey = true
+	body := evolutionField(102, "body", schemapb.DataType_Text, true)
+	body.TypeParams = []*commonpb.KeyValuePair{{Key: common.EnableAnalyzerKey, Value: "true"}}
+	vector := evolutionField(103, "embedding", schemapb.DataType_FloatVector, false)
+	vector.TypeParams = []*commonpb.KeyValuePair{{Key: common.DimKey, Value: "128"}}
+	title := evolutionField(104, "title", schemapb.DataType_VarChar, true)
+	title.TypeParams = []*commonpb.KeyValuePair{{Key: common.MaxLengthKey, Value: "128"}}
+	tags := evolutionField(105, "tags", schemapb.DataType_Array, true)
+	tags.ElementType = schemapb.DataType_VarChar
+	tags.TypeParams = []*commonpb.KeyValuePair{
+		{Key: common.MaxCapacityKey, Value: "64"},
+		{Key: common.MaxLengthKey, Value: "32"},
+	}
+	return &schemapb.CollectionSchema{
+		Name:       "evolution",
+		Fields:     []*schemapb.FieldSchema{rowID, timestamp, pk, body, vector, title, tags},
+		Properties: []*commonpb.KeyValuePair{{Key: common.MaxFieldIDKey, Value: "105"}},
+	}
+}
+
+func evolutionSchemaWithFunction() *schemapb.CollectionSchema {
+	schema := evolutionBaseSchema()
+	output := evolutionField(106, "sparse", schemapb.DataType_SparseFloatVector, false)
+	output.IsFunctionOutput = true
+	schema.Fields = append(schema.Fields, output)
+	schema.Functions = []*schemapb.FunctionSchema{evolutionFunction(200, 102, 106)}
+	setMaxFieldID(schema, 106)
+	return schema
+}
+
+func evolutionSchemaWithStruct() *schemapb.CollectionSchema {
+	schema := evolutionBaseSchema()
+	child := evolutionField(107, "profile[values]", schemapb.DataType_Array, true)
+	child.ElementType = schemapb.DataType_Int64
+	child.TypeParams = []*commonpb.KeyValuePair{{Key: common.MaxCapacityKey, Value: "32"}}
+	schema.StructArrayFields = []*schemapb.StructArrayFieldSchema{
+		{
+			FieldID:  106,
+			Name:     "profile",
+			Nullable: true,
+			TypeParams: []*commonpb.KeyValuePair{
+				{Key: common.MaxCapacityKey, Value: "64"},
+			},
+			Fields: []*schemapb.FieldSchema{child},
+		},
+	}
+	setMaxFieldID(schema, 107)
+	return schema
+}
+
+func evolutionFunction(id, inputID, outputID int64) *schemapb.FunctionSchema {
+	return &schemapb.FunctionSchema{
+		Id:               id,
+		Name:             "bm25",
+		Type:             schemapb.FunctionType_BM25,
+		InputFieldIds:    []int64{inputID},
+		InputFieldNames:  []string{"body"},
+		OutputFieldIds:   []int64{outputID},
+		OutputFieldNames: []string{"sparse"},
+	}
+}
+
+func evolutionField(id int64, name string, dataType schemapb.DataType, nullable bool) *schemapb.FieldSchema {
+	return &schemapb.FieldSchema{FieldID: id, Name: name, DataType: dataType, Nullable: nullable}
+}
+
+func evolutionFieldByID(schema *schemapb.CollectionSchema, fieldID int64) *schemapb.FieldSchema {
+	for _, field := range schema.GetFields() {
+		if field.GetFieldID() == fieldID {
+			return field
+		}
+	}
+	return nil
+}
+
+func removeEvolutionField(fields []*schemapb.FieldSchema, fieldID int64) []*schemapb.FieldSchema {
+	result := make([]*schemapb.FieldSchema, 0, len(fields))
+	for _, field := range fields {
+		if field.GetFieldID() != fieldID {
+			result = append(result, field)
+		}
+	}
+	return result
+}
+
+func setEvolutionTypeParam(field *schemapb.FieldSchema, key, value string) {
+	for _, param := range field.GetTypeParams() {
+		if param.GetKey() == key {
+			param.Value = value
+			return
+		}
+	}
+	field.TypeParams = append(field.TypeParams, &commonpb.KeyValuePair{Key: key, Value: value})
+}
+
+func removeEvolutionTypeParam(field *schemapb.FieldSchema, key string) {
+	params := make([]*commonpb.KeyValuePair, 0, len(field.GetTypeParams()))
+	for _, param := range field.GetTypeParams() {
+		if param.GetKey() != key {
+			params = append(params, param)
+		}
+	}
+	field.TypeParams = params
+}
+
+func setMaxFieldID(schema *schemapb.CollectionSchema, value int64) {
+	for _, property := range schema.GetProperties() {
+		if property.GetKey() == common.MaxFieldIDKey {
+			property.Value = strconv.FormatInt(value, 10)
+			return
+		}
+	}
+	schema.Properties = append(schema.Properties, &commonpb.KeyValuePair{
+		Key:   common.MaxFieldIDKey,
+		Value: strconv.FormatInt(value, 10),
+	})
+}
+
+func removeEvolutionProperty(schema *schemapb.CollectionSchema, key string) {
+	properties := make([]*commonpb.KeyValuePair, 0, len(schema.GetProperties()))
+	for _, property := range schema.GetProperties() {
+		if property.GetKey() != key {
+			properties = append(properties, property)
+		}
+	}
+	schema.Properties = properties
+}
+
+func setEvolutionProperty(schema *schemapb.CollectionSchema, key, value string) {
+	for _, property := range schema.GetProperties() {
+		if property.GetKey() == key {
+			property.Value = value
+			return
+		}
+	}
+	schema.Properties = append(schema.Properties, &commonpb.KeyValuePair{Key: key, Value: value})
+}