enhance: route partition namespace requests (#50598)

issue: #50748

## Summary
- route namespace requests to partition names when
`namespace.mode=partition`
- keep the default namespace mode using `$namespace_id` plan filtering
- prevent altering `namespace.mode` after collection creation

Signed-off-by: sunby <sunbingyi1992@gmail.com>

internal/proxy/rate_limit_interceptor.go

@@ -88,6 +88,25 @@ type reqCollName interface {
 	requestutil.CollectionNameGetter
 }
 
+func getRequestNamespace(req any) *string {
+	switch r := req.(type) {
+	case *milvuspb.InsertRequest:
+		return r.Namespace
+	case *milvuspb.UpsertRequest:
+		return r.Namespace
+	case *milvuspb.DeleteRequest:
+		return r.Namespace
+	case *milvuspb.SearchRequest:
+		return r.Namespace
+	case *milvuspb.HybridSearchRequest:
+		return r.Namespace
+	case *milvuspb.QueryRequest:
+		return r.Namespace
+	default:
+		return nil
+	}
+}
+
 func getCollectionAndPartitionID(ctx context.Context, r reqPartName) (int64, map[int64][]int64, error) {
 	db, err := globalMetaCache.GetDatabaseInfo(ctx, r.GetDbName())
 	if err != nil {
@@ -97,11 +116,33 @@ func getCollectionAndPartitionID(ctx context.Context, r reqPartName) (int64, map
 	if err != nil {
 		return 0, nil, err
 	}
-	if r.GetPartitionName() == "" {
-		collectionSchema, err := globalMetaCache.GetCollectionSchema(ctx, r.GetDbName(), r.GetCollectionName())
+
+	var collectionSchema *schemaInfo
+	if namespace := getRequestNamespace(r); namespace != nil {
+		collectionSchema, err = globalMetaCache.GetCollectionSchema(ctx, r.GetDbName(), r.GetCollectionName())
 		if err != nil {
 			return 0, nil, err
 		}
+		partitionName, namespaceAsPartition, err := resolveNamespacePartitionName(collectionSchema.CollectionSchema, namespace, r.GetPartitionName())
+		if err != nil {
+			return 0, nil, err
+		}
+		if namespaceAsPartition {
+			part, err := globalMetaCache.GetPartitionInfo(ctx, r.GetDbName(), r.GetCollectionName(), partitionName)
+			if err != nil {
+				return 0, nil, err
+			}
+			return db.dbID, map[int64][]int64{collectionID: {part.partitionID}}, nil
+		}
+	}
+
+	if r.GetPartitionName() == "" {
+		if collectionSchema == nil {
+			collectionSchema, err = globalMetaCache.GetCollectionSchema(ctx, r.GetDbName(), r.GetCollectionName())
+			if err != nil {
+				return 0, nil, err
+			}
+		}
 		if collectionSchema.IsPartitionKeyCollection() {
 			return db.dbID, map[int64][]int64{collectionID: {}}, nil
 		}
@@ -122,8 +163,20 @@ func getCollectionAndPartitionIDs(ctx context.Context, r reqPartNames) (int64, m
 	if err != nil {
 		return 0, nil, err
 	}
-	parts := make([]int64, len(r.GetPartitionNames()))
-	for i, s := range r.GetPartitionNames() {
+	partitionNames := r.GetPartitionNames()
+	if namespace := getRequestNamespace(r); namespace != nil {
+		collectionSchema, err := globalMetaCache.GetCollectionSchema(ctx, r.GetDbName(), r.GetCollectionName())
+		if err != nil {
+			return 0, nil, err
+		}
+		partitionNames, _, err = resolveNamespacePartitionNames(collectionSchema.CollectionSchema, namespace, partitionNames)
+		if err != nil {
+			return 0, nil, err
+		}
+	}
+
+	parts := make([]int64, len(partitionNames))
+	for i, s := range partitionNames {
 		part, err := globalMetaCache.GetPartitionInfo(ctx, r.GetDbName(), r.GetCollectionName(), s)
 		if err != nil {
 			return 0, nil, err

internal/proxy/rate_limit_interceptor_test.go

@@ -28,6 +28,8 @@ 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/pkg/v3/common"
 	"github.com/milvus-io/milvus/pkg/v3/proto/internalpb"
 	"github.com/milvus-io/milvus/pkg/v3/util"
 	"github.com/milvus-io/milvus/pkg/v3/util/merr"
@@ -264,6 +266,77 @@ func TestRateLimitInterceptor(t *testing.T) {
 		assert.NoError(t, err)
 	})
 
+	t.Run("namespace partition mode request info", func(t *testing.T) {
+		namespace := "tenant_partition"
+		schema := &schemapb.CollectionSchema{
+			EnableNamespace: true,
+			Properties: []*commonpb.KeyValuePair{
+				{Key: common.NamespaceModeKey, Value: common.NamespaceModePartition},
+			},
+		}
+		mockCache := NewMockCache(t)
+		mockCache.EXPECT().GetDatabaseInfo(mock.Anything, mock.Anything).Return(&databaseInfo{
+			dbID:             100,
+			createdTimestamp: 1,
+		}, nil).Times(4)
+		mockCache.EXPECT().GetCollectionID(mock.Anything, mock.Anything, mock.Anything).Return(int64(1), nil).Times(4)
+		mockCache.EXPECT().GetCollectionSchema(mock.Anything, mock.Anything, mock.Anything).Return(newSchemaInfo(schema), nil).Times(4)
+		mockCache.EXPECT().GetPartitionInfo(mock.Anything, mock.Anything, mock.Anything, namespace).Return(&partitionInfo{
+			name:                namespace,
+			partitionID:         20,
+			createdTimestamp:    10001,
+			createdUtcTimestamp: 10002,
+		}, nil).Times(4)
+		globalMetaCache = mockCache
+
+		database, col2part, rt, _, err := GetRequestInfo(context.Background(), &milvuspb.InsertRequest{
+			CollectionName: "foo",
+			DbName:         "db1",
+			Namespace:      &namespace,
+		})
+		assert.NoError(t, err)
+		assert.Equal(t, int64(100), database)
+		assert.Equal(t, internalpb.RateType_DMLInsert, rt)
+		assert.Equal(t, []int64{20}, col2part[1])
+
+		database, col2part, rt, _, err = GetRequestInfo(context.Background(), &milvuspb.DeleteRequest{
+			CollectionName: "foo",
+			DbName:         "db1",
+			Namespace:      &namespace,
+		})
+		assert.NoError(t, err)
+		assert.Equal(t, int64(100), database)
+		assert.Equal(t, internalpb.RateType_DMLDelete, rt)
+		assert.Equal(t, []int64{20}, col2part[1])
+
+		database, col2part, rt, size, err := GetRequestInfo(context.Background(), &milvuspb.SearchRequest{
+			CollectionName: "foo",
+			DbName:         "db1",
+			Nq:             5,
+			Namespace:      &namespace,
+		})
+		assert.NoError(t, err)
+		assert.Equal(t, int64(100), database)
+		assert.Equal(t, internalpb.RateType_DQLSearch, rt)
+		assert.Equal(t, 5, size)
+		assert.Equal(t, []int64{20}, col2part[1])
+
+		database, col2part, rt, size, err = GetRequestInfo(context.Background(), &milvuspb.HybridSearchRequest{
+			CollectionName: "foo",
+			DbName:         "db1",
+			Namespace:      &namespace,
+			Requests: []*milvuspb.SearchRequest{
+				{Nq: 2},
+				{Nq: 3},
+			},
+		})
+		assert.NoError(t, err)
+		assert.Equal(t, int64(100), database)
+		assert.Equal(t, internalpb.RateType_DQLSearch, rt)
+		assert.Equal(t, 5, size)
+		assert.Equal(t, []int64{20}, col2part[1])
+	})
+
 	t.Run("test GetFailedResponse", func(t *testing.T) {
 		testGetFailedResponse := func(req interface{}, rt internalpb.RateType, err error, fullMethod string) {
 			rsp := GetFailedResponse(req, err)

internal/proxy/search_pipeline.go

@@ -1341,6 +1341,7 @@ type requeryOperator struct {
 	consistencyLevel   commonpb.ConsistencyLevel
 	guaranteeTimestamp uint64
 	namespace          *string
+	planNamespace      *string
 
 	node types.ProxyComponent
 }
@@ -1392,6 +1393,7 @@ func newRequeryOperator(t *searchTask, _ map[string]any) (operator, error) {
 		partitionIDs:       t.GetPartitionIDs(),
 		node:               t.node,
 		namespace:          t.request.Namespace,
+		planNamespace:      namespaceForPlan(t.schema.CollectionSchema, t.request.Namespace),
 	}, nil
 }
 
@@ -1429,7 +1431,7 @@ func (op *requeryOperator) requery(ctx context.Context, span trace.Span, ids *sc
 		Namespace:             op.namespace,
 	}
 	plan := planparserv2.CreateRequeryPlan(op.primaryFieldSchema, ids)
-	plan.Namespace = op.namespace
+	plan.Namespace = op.planNamespace
 	channelsMvcc := make(map[string]Timestamp)
 	for k, v := range op.queryChannelsTs {
 		channelsMvcc[k] = v

internal/proxy/search_util.go

@@ -1079,6 +1079,7 @@ func convertHybridSearchToSearch(req *milvuspb.HybridSearchRequest) *milvuspb.Se
 		SearchByPrimaryKeys:   false,
 		SubReqs:               nil,
 		FunctionScore:         req.FunctionScore,
+		Namespace:             req.Namespace,
 	}
 
 	for _, sub := range req.GetRequests() {

internal/proxy/task_delete.go

@@ -294,6 +294,13 @@ func (dr *deleteRunner) Init(ctx context.Context) error {
 	if err := validateTextStorageV3Enabled(dr.schema.CollectionSchema); err != nil {
 		return ErrWithLog(log, "TEXT field requires StorageV3", err)
 	}
+	if namespacePartitionModeEnabled(dr.schema.CollectionSchema) {
+		partitionName, _, err := resolveNamespacePartitionName(dr.schema.CollectionSchema, dr.req.Namespace, dr.req.GetPartitionName())
+		if err != nil {
+			return err
+		}
+		dr.req.PartitionName = partitionName
+	}
 
 	colInfo, err := globalMetaCache.GetCollectionInfo(ctx, dr.req.GetDbName(), collName, dr.collectionID)
 	if err != nil {

internal/proxy/task_delete_test.go

@@ -11,6 +11,7 @@ import (
 	"github.com/stretchr/testify/suite"
 	"google.golang.org/grpc"
 
+	"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/allocator"
@@ -415,6 +416,53 @@ func (s *DeleteRunnerSuite) TestInitSuccess() {
 		s.Equal(1, len(dr.partitionIDs))
 		s.EqualValues(1000, dr.partitionIDs[0])
 	})
+
+	s.Run("namespace partition mode routes to namespace partition", func() {
+		mockChMgr := NewMockChannelsMgr(s.T())
+		namespace := "tenant_partition"
+		dr := deleteRunner{
+			req: &milvuspb.DeleteRequest{
+				CollectionName: s.collectionName,
+				Namespace:      &namespace,
+				Expr:           "pk == 1",
+			},
+			chMgr: mockChMgr,
+		}
+		s.mockCache.EXPECT().GetDatabaseInfo(mock.Anything, mock.Anything).Return(&databaseInfo{dbID: 0}, nil)
+		s.mockCache.EXPECT().GetCollectionID(mock.Anything, mock.Anything, mock.Anything).Return(s.collectionID, nil)
+		s.mockCache.EXPECT().GetCollectionInfo(mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(&collectionInfo{}, nil)
+		schema := &schemapb.CollectionSchema{
+			Name:            s.collectionName,
+			EnableNamespace: true,
+			Properties: []*commonpb.KeyValuePair{
+				{Key: common.NamespaceModeKey, Value: common.NamespaceModePartition},
+			},
+			Fields: []*schemapb.FieldSchema{
+				{
+					FieldID:      common.StartOfUserFieldID,
+					Name:         "pk",
+					IsPrimaryKey: true,
+					DataType:     schemapb.DataType_Int64,
+				},
+				{
+					FieldID:  common.StartOfUserFieldID + 1,
+					Name:     "non_pk",
+					DataType: schemapb.DataType_Int64,
+				},
+			},
+		}
+		schemaInfo := newSchemaInfo(schema)
+		s.mockCache.EXPECT().GetCollectionSchema(mock.Anything, mock.Anything, mock.Anything).Return(schemaInfo, nil).Once()
+		s.mockCache.EXPECT().GetPartitionID(mock.Anything, mock.Anything, mock.Anything, namespace).Return(int64(1000), nil)
+		mockChMgr.EXPECT().getVChannels(mock.Anything).Return([]string{"vchan1"}, nil)
+
+		globalMetaCache = s.mockCache
+		s.NoError(dr.Init(context.Background()))
+
+		s.Equal(namespace, dr.req.GetPartitionName())
+		s.Equal([]UniqueID{1000}, dr.partitionIDs)
+		s.Nil(dr.plan.Namespace)
+	})
 }
 
 func (s *DeleteRunnerSuite) TestInitFailure() {

internal/proxy/task_query.go

@@ -700,10 +700,13 @@ func (t *queryTask) PreExecute(ctx context.Context) error {
 	if err := validateTextStorageV3Enabled(t.schema.CollectionSchema); err != nil {
 		return err
 	}
-	err = common.CheckNamespace(t.schema.CollectionSchema, t.request.Namespace)
+	partitionNames, namespaceAsPartition, err := resolveNamespacePartitionNames(t.schema.CollectionSchema, t.request.Namespace, t.request.GetPartitionNames())
 	if err != nil {
 		return err
 	}
+	if namespaceAsPartition {
+		t.request.PartitionNames = partitionNames
+	}
 
 	t.partitionKeyMode, err = isPartitionKeyMode(ctx, t.request.GetDbName(), collectionName)
 	if err != nil {
@@ -826,7 +829,7 @@ func (t *queryTask) PreExecute(ctx context.Context) error {
 	if t.hasCountStar() && t.queryParams.limit != typeutil.Unlimited && len(t.GetGroupByFieldIds()) == 0 {
 		return merr.WrapErrParameterInvalidMsg("count entities with pagination is not allowed")
 	}
-	t.plan.Namespace = t.request.Namespace
+	t.plan.Namespace = namespaceForPlan(t.schema.CollectionSchema, t.request.Namespace)
 
 	t.SerializedExprPlan, err = proto.Marshal(t.plan)
 	if err != nil {

internal/proxy/task_search.go

@@ -196,6 +196,13 @@ func (t *searchTask) PreExecute(ctx context.Context) error {
 		log.Warn(ctx, "is partition key mode failed", mlog.Err(err))
 		return err
 	}
+	partitionNames, namespaceAsPartition, err := resolveNamespacePartitionNames(t.schema.CollectionSchema, t.request.Namespace, t.request.GetPartitionNames())
+	if err != nil {
+		return err
+	}
+	if namespaceAsPartition {
+		t.request.PartitionNames = partitionNames
+	}
 	if t.partitionKeyMode && len(t.request.GetPartitionNames()) != 0 {
 		return merr.WrapErrParameterInvalidMsg("not support manually specifying the partition names if partition key mode is used")
 	}
@@ -212,10 +219,6 @@ func (t *searchTask) PreExecute(ctx context.Context) error {
 			return err
 		}
 	}
-	err = common.CheckNamespace(t.schema.CollectionSchema, t.request.Namespace)
-	if err != nil {
-		return err
-	}
 
 	var aggs []agg.AggregateBase
 	t.translatedOutputFields, t.userOutputFields, t.userDynamicFields, aggs, t.userRequestedPkFieldExplicitly, err = translateOutputFields(t.request.OutputFields, t.schema, true)
@@ -678,7 +681,7 @@ func (t *searchTask) initAdvancedSearchRequest(ctx context.Context) error {
 			plan.OutputFieldIds = allFieldIDs.Collect()
 			plan.DynamicFields = t.userDynamicFields
 		}
-		plan.Namespace = t.request.Namespace
+		plan.Namespace = namespaceForPlan(t.schema.CollectionSchema, t.request.Namespace)
 
 		internalSubReq.SerializedExprPlan, err = proto.Marshal(plan)
 		if err != nil {
@@ -982,7 +985,7 @@ func (t *searchTask) initSearchRequest(ctx context.Context) error {
 			}
 		}
 	}
-	plan.Namespace = t.request.Namespace
+	plan.Namespace = namespaceForPlan(t.schema.CollectionSchema, t.request.Namespace)
 
 	// Propagate agg-path overrides into queryInfo BEFORE plan serialization so
 	// segcore sees the derived topK / groupSize and plural GroupByFieldIds.
@@ -1094,7 +1097,7 @@ func (t *searchTask) initSearchRequest(ctx context.Context) error {
 func (t *searchTask) skipRequeryByNamespacePartitionMode() bool {
 	return t.schema != nil &&
 		t.schema.CollectionSchema != nil &&
-		common.IsNamespaceModePartition(t.schema.GetProperties()...)
+		namespacePartitionModeEnabled(t.schema.CollectionSchema)
 }
 
 // convertPlaceholderIfNeeded converts fp32 vectors to fp16/bf16 if the target field uses lower precision.

internal/proxy/task_search_test.go

@@ -567,7 +567,8 @@ func TestSearchTask_NamespacePartitionModeSkipsRequery(t *testing.T) {
 	ctx := context.Background()
 
 	schema := &schemapb.CollectionSchema{
-		Name: "test_collection",
+		Name:            "test_collection",
+		EnableNamespace: true,
 		Properties: []*commonpb.KeyValuePair{
 			{Key: common.NamespaceModeKey, Value: common.NamespaceModePartition},
 		},
@@ -578,6 +579,14 @@ func TestSearchTask_NamespacePartitionModeSkipsRequery(t *testing.T) {
 	}
 	schemaInfo := newSchemaInfo(schema)
 
+	t.Run("namespace disabled does not skip", func(t *testing.T) {
+		disabledSchema := proto.Clone(schema).(*schemapb.CollectionSchema)
+		disabledSchema.EnableNamespace = false
+		task := &searchTask{schema: newSchemaInfo(disabledSchema)}
+
+		require.False(t, task.skipRequeryByNamespacePartitionMode())
+	})
+
 	makePlaceholderGroup := func() []byte {
 		phg := &commonpb.PlaceholderGroup{
 			Placeholders: []*commonpb.PlaceholderValue{{

internal/proxy/task_upsert.go

@@ -203,7 +203,7 @@ func retrieveByPKs(ctx context.Context, t *upsertTask, ids *schemapb.IDs, output
 	}
 
 	plan := planparserv2.CreateRequeryPlan(pkField, ids)
-	plan.Namespace = t.req.Namespace
+	plan.Namespace = namespaceForPlan(t.schema.CollectionSchema, t.req.Namespace)
 	qt := &queryTask{
 		ctx:       t.ctx,
 		Condition: NewTaskCondition(t.ctx),
@@ -1304,10 +1304,13 @@ func (it *upsertTask) PreExecute(ctx context.Context) error {
 		it.req.PartialUpdate = true
 	}
 
-	err = common.CheckNamespace(schema.CollectionSchema, it.req.Namespace)
+	partitionName, namespaceAsPartition, err := resolveNamespacePartitionName(schema.CollectionSchema, it.req.Namespace, it.req.GetPartitionName())
 	if err != nil {
 		return err
 	}
+	if namespaceAsPartition {
+		it.req.PartitionName = partitionName
+	}
 
 	if it.req.GetPartialUpdate() && len(schema.GetStructArrayFields()) > 0 {
 		return merr.WrapErrParameterInvalidMsg("partial upsert is not supported for collections with struct array fields")

internal/proxy/util.go

@@ -2745,14 +2745,68 @@ func checkDynamicFieldDataForPartialUpdate(schema *schemapb.CollectionSchema, in
 	return doCheckDynamicFieldData(schema, insertMsg, true)
 }
 
+func namespacePartitionModeEnabled(schema *schemapb.CollectionSchema) bool {
+	return schema.GetEnableNamespace() && common.IsNamespaceModePartition(schema.GetProperties()...)
+}
+
+func resolveNamespacePartitionName(schema *schemapb.CollectionSchema, namespace *string, partitionName string) (string, bool, error) {
+	if err := common.CheckNamespace(schema, namespace); err != nil {
+		return "", false, err
+	}
+	if !namespacePartitionModeEnabled(schema) {
+		return partitionName, false, nil
+	}
+
+	namespacePartitionName := *namespace
+	if err := validatePartitionTag(namespacePartitionName, true); err != nil {
+		return "", true, err
+	}
+	if partitionName != "" && partitionName != namespacePartitionName {
+		return "", true, merr.WrapErrParameterInvalidMsg("partition name %q mismatches namespace %q", partitionName, namespacePartitionName)
+	}
+	return namespacePartitionName, true, nil
+}
+
+func resolveNamespacePartitionNames(schema *schemapb.CollectionSchema, namespace *string, partitionNames []string) ([]string, bool, error) {
+	if err := common.CheckNamespace(schema, namespace); err != nil {
+		return nil, false, err
+	}
+	if !namespacePartitionModeEnabled(schema) {
+		return partitionNames, false, nil
+	}
+
+	namespacePartitionName := *namespace
+	if err := validatePartitionTag(namespacePartitionName, true); err != nil {
+		return nil, true, err
+	}
+	if len(partitionNames) == 0 {
+		return []string{namespacePartitionName}, true, nil
+	}
+	if len(partitionNames) == 1 && partitionNames[0] == namespacePartitionName {
+		return partitionNames, true, nil
+	}
+	return nil, true, merr.WrapErrParameterInvalidMsg("partition names %v mismatch namespace %q", partitionNames, namespacePartitionName)
+}
+
+func namespaceForPlan(schema *schemapb.CollectionSchema, namespace *string) *string {
+	if namespacePartitionModeEnabled(schema) {
+		return nil
+	}
+	return namespace
+}
+
 func addNamespaceData(schema *schemapb.CollectionSchema, insertMsg *msgstream.InsertMsg) error {
-	err := common.CheckNamespace(schema, insertMsg.Namespace)
+	partitionName, namespaceAsPartition, err := resolveNamespacePartitionName(schema, insertMsg.Namespace, insertMsg.GetPartitionName())
 	if err != nil {
 		return err
 	}
 	if !schema.GetEnableNamespace() {
 		return nil
 	}
+	if namespaceAsPartition {
+		insertMsg.PartitionName = partitionName
+		return nil
+	}
 
 	// check namespace field exists
 	namespaceField := typeutil.GetFieldByName(schema, common.NamespaceFieldName)
@@ -2926,6 +2980,13 @@ func GetRequestInfo(ctx context.Context, req proto.Message) (int64, map[int64][]
 	case *milvuspb.SearchRequest:
 		dbID, collToPartIDs, err := getCollectionAndPartitionIDs(ctx, req.(reqPartNames))
 		return dbID, collToPartIDs, internalpb.RateType_DQLSearch, int(r.GetNq()), err
+	case *milvuspb.HybridSearchRequest:
+		dbID, collToPartIDs, err := getCollectionAndPartitionIDs(ctx, req.(reqPartNames))
+		nq := 0
+		for _, subReq := range r.GetRequests() {
+			nq += int(subReq.GetNq())
+		}
+		return dbID, collToPartIDs, internalpb.RateType_DQLSearch, nq, err
 	case *milvuspb.QueryRequest:
 		dbID, collToPartIDs, err := getCollectionAndPartitionIDs(ctx, req.(reqPartNames))
 		return dbID, collToPartIDs, internalpb.RateType_DQLQuery, 1, err // think of the query request's nq as 1

internal/proxy/util_test.go

@@ -189,6 +189,111 @@ func TestValidateResourceGroupName(t *testing.T) {
 	}
 }
 
+func TestNamespacePartitionRoutingHelpers(t *testing.T) {
+	namespace := "tenant_partition"
+	partitionModeSchema := &schemapb.CollectionSchema{
+		EnableNamespace: true,
+		Properties: []*commonpb.KeyValuePair{
+			{Key: common.NamespaceModeKey, Value: common.NamespaceModePartition},
+		},
+	}
+	partitionKeyModeSchema := &schemapb.CollectionSchema{
+		EnableNamespace: true,
+		Fields: []*schemapb.FieldSchema{
+			{FieldID: 101, Name: common.NamespaceFieldName, DataType: schemapb.DataType_VarChar, IsPartitionKey: true},
+		},
+	}
+
+	t.Run("single partition name", func(t *testing.T) {
+		partitionName, usedNamespacePartition, err := resolveNamespacePartitionName(partitionModeSchema, &namespace, "")
+		require.NoError(t, err)
+		assert.True(t, usedNamespacePartition)
+		assert.Equal(t, namespace, partitionName)
+
+		partitionName, usedNamespacePartition, err = resolveNamespacePartitionName(partitionModeSchema, &namespace, namespace)
+		require.NoError(t, err)
+		assert.True(t, usedNamespacePartition)
+		assert.Equal(t, namespace, partitionName)
+
+		_, _, err = resolveNamespacePartitionName(partitionModeSchema, &namespace, "other_partition")
+		require.ErrorIs(t, err, merr.ErrParameterInvalid)
+
+		emptyNamespace := ""
+		_, _, err = resolveNamespacePartitionName(partitionModeSchema, &emptyNamespace, "")
+		require.ErrorIs(t, err, merr.ErrParameterInvalid)
+	})
+
+	t.Run("partition name list", func(t *testing.T) {
+		partitionNames, usedNamespacePartition, err := resolveNamespacePartitionNames(partitionModeSchema, &namespace, nil)
+		require.NoError(t, err)
+		assert.True(t, usedNamespacePartition)
+		assert.Equal(t, []string{namespace}, partitionNames)
+
+		partitionNames, usedNamespacePartition, err = resolveNamespacePartitionNames(partitionModeSchema, &namespace, []string{namespace})
+		require.NoError(t, err)
+		assert.True(t, usedNamespacePartition)
+		assert.Equal(t, []string{namespace}, partitionNames)
+
+		_, _, err = resolveNamespacePartitionNames(partitionModeSchema, &namespace, []string{"other_partition"})
+		require.ErrorIs(t, err, merr.ErrParameterInvalid)
+
+		_, _, err = resolveNamespacePartitionNames(partitionModeSchema, &namespace, []string{namespace, "other_partition"})
+		require.ErrorIs(t, err, merr.ErrParameterInvalid)
+	})
+
+	t.Run("default mode keeps namespace as plan filter", func(t *testing.T) {
+		partitionName, usedNamespacePartition, err := resolveNamespacePartitionName(partitionKeyModeSchema, &namespace, "")
+		require.NoError(t, err)
+		assert.False(t, usedNamespacePartition)
+		assert.Empty(t, partitionName)
+
+		partitionNames, usedNamespacePartition, err := resolveNamespacePartitionNames(partitionKeyModeSchema, &namespace, nil)
+		require.NoError(t, err)
+		assert.False(t, usedNamespacePartition)
+		assert.Empty(t, partitionNames)
+
+		assert.Same(t, &namespace, namespaceForPlan(partitionKeyModeSchema, &namespace))
+		assert.Nil(t, namespaceForPlan(partitionModeSchema, &namespace))
+	})
+}
+
+func TestAddNamespaceDataPartitionMode(t *testing.T) {
+	namespace := "tenant_partition"
+	schema := &schemapb.CollectionSchema{
+		EnableNamespace: true,
+		Properties: []*commonpb.KeyValuePair{
+			{Key: common.NamespaceModeKey, Value: common.NamespaceModePartition},
+		},
+	}
+	insertMsg := &msgstream.InsertMsg{
+		InsertRequest: &msgpb.InsertRequest{
+			Namespace: &namespace,
+			NumRows:   3,
+		},
+	}
+
+	err := addNamespaceData(schema, insertMsg)
+	require.NoError(t, err)
+	assert.Equal(t, namespace, insertMsg.GetPartitionName())
+	assert.Empty(t, insertMsg.GetFieldsData())
+}
+
+func TestConvertHybridSearchToSearchCopiesNamespace(t *testing.T) {
+	namespace := "tenant_partition"
+	req := &milvuspb.HybridSearchRequest{
+		DbName:         "default",
+		CollectionName: "coll",
+		Namespace:      &namespace,
+		Requests: []*milvuspb.SearchRequest{
+			{Dsl: "pk > 0"},
+		},
+	}
+
+	searchReq := convertHybridSearchToSearch(req)
+	require.NotNil(t, searchReq.Namespace)
+	assert.Equal(t, namespace, searchReq.GetNamespace())
+}
+
 func TestValidateDatabaseName(t *testing.T) {
 	assert.Nil(t, ValidateDatabaseName("dbname"))
 	assert.Nil(t, ValidateDatabaseName("_123abc"))

internal/rootcoord/create_collection_task.go

@@ -421,6 +421,10 @@ func (t *createCollectionTask) handleNamespaceField(ctx context.Context, schema
 		return merr.WrapErrParameterInvalidMsg("namespace is not supported with partition key mode")
 	}
 
+	if common.IsNamespaceModePartition(t.Req.GetProperties()...) {
+		return nil
+	}
+
 	schema.Fields = append(schema.Fields, &schemapb.FieldSchema{
 		Name:           common.NamespaceFieldName,
 		IsPartitionKey: true,

internal/rootcoord/create_collection_task_test.go

@@ -2420,6 +2420,27 @@ func TestNamespaceProperty(t *testing.T) {
 		assert.True(t, hasNamespaceField(schema))
 	})
 
+	t.Run("test namespace partition mode", func(t *testing.T) {
+		schema := initSchema()
+		task := &createCollectionTask{
+			Req: &milvuspb.CreateCollectionRequest{
+				CollectionName: collectionName,
+				Properties: []*commonpb.KeyValuePair{
+					{Key: common.NamespaceModeKey, Value: common.NamespaceModePartition},
+				},
+			},
+			header: &message.CreateCollectionMessageHeader{},
+			body: &message.CreateCollectionRequest{
+				CollectionSchema: schema,
+			},
+		}
+
+		err := task.handleNamespaceField(ctx, schema)
+		assert.NoError(t, err)
+		assert.False(t, hasNamespaceField(schema))
+		assert.False(t, hasIsolationProperty(task.Req.Properties...))
+	})
+
 	t.Run("test namespace disabled with isolation and partition key", func(t *testing.T) {
 		schema := initSchema()
 		schema.EnableNamespace = false

internal/rootcoord/ddl_callbacks_alter_collection_properties.go

@@ -2,6 +2,7 @@ package rootcoord
 
 import (
 	"context"
+	"strings"
 
 	"github.com/cockroachdb/errors"
 	"github.com/samber/lo"
@@ -51,6 +52,10 @@ func (c *Core) broadcastAlterCollectionForAlterCollection(ctx context.Context, r
 		return err
 	}
 
+	if err := validateNamespaceModeImmutable(req.GetProperties(), req.GetDeleteKeys()); err != nil {
+		return err
+	}
+
 	if funcutil.SliceContain(req.GetDeleteKeys(), common.EnableDynamicSchemaKey) {
 		return merr.WrapErrParameterInvalidMsg("cannot delete key %s, dynamic field schema could support set to true/false", common.EnableDynamicSchemaKey)
 	}
@@ -207,6 +212,26 @@ func (c *Core) broadcastAlterCollectionForAlterCollection(ctx context.Context, r
 	return nil
 }
 
+func validateNamespaceModeImmutable(properties []*commonpb.KeyValuePair, deleteKeys []string) error {
+	for _, prop := range properties {
+		if prop.GetKey() == common.NamespaceModeKey {
+			return merr.WrapErrParameterInvalidMsg("cannot alter %s via alter_collection_properties; namespace mode is immutable after collection creation", common.NamespaceModeKey)
+		}
+		if strings.EqualFold(prop.GetKey(), common.NamespaceModeKey) {
+			return merr.WrapErrParameterInvalidMsg("invalid property key %q, did you mean %q?", prop.GetKey(), common.NamespaceModeKey)
+		}
+	}
+	for _, key := range deleteKeys {
+		if key == common.NamespaceModeKey {
+			return merr.WrapErrParameterInvalidMsg("cannot delete %s; namespace mode is immutable after collection creation", common.NamespaceModeKey)
+		}
+		if strings.EqualFold(key, common.NamespaceModeKey) {
+			return merr.WrapErrParameterInvalidMsg("invalid property key %q, did you mean %q?", key, common.NamespaceModeKey)
+		}
+	}
+	return nil
+}
+
 // broadcastAlterCollectionForAlterDynamicField broadcasts the put collection message for alter dynamic field.
 func (c *Core) broadcastAlterCollectionForAlterDynamicField(ctx context.Context, req *milvuspb.AlterCollectionRequest, targetValue bool) error {
 	if len(req.GetProperties()) != 1 {

internal/rootcoord/ddl_callbacks_alter_collection_properties_test.go

@@ -129,6 +129,34 @@ func TestDDLCallbacksAlterCollectionProperties(t *testing.T) {
 	// atler a property of a collection.
 	createCollectionAndAliasForTest(t, ctx, core, dbName, collectionName)
 	assertReplicaNumber(t, ctx, core, dbName, collectionName, 1)
+
+	for _, tc := range []struct {
+		name       string
+		properties []*commonpb.KeyValuePair
+		deleteKeys []string
+	}{
+		{
+			name:       "set namespace mode",
+			properties: []*commonpb.KeyValuePair{{Key: common.NamespaceModeKey, Value: common.NamespaceModePartition}},
+		},
+		{
+			name:       "delete namespace mode",
+			deleteKeys: []string{common.NamespaceModeKey},
+		},
+	} {
+		t.Run("reject "+tc.name, func(t *testing.T) {
+			resp, err = core.AlterCollection(ctx, &milvuspb.AlterCollectionRequest{
+				DbName:         dbName,
+				CollectionName: collectionName,
+				Properties:     tc.properties,
+				DeleteKeys:     tc.deleteKeys,
+			})
+			alterErr := merr.CheckRPCCall(resp, err)
+			require.ErrorIs(t, alterErr, merr.ErrParameterInvalid)
+			require.ErrorContains(t, alterErr, common.NamespaceModeKey)
+		})
+	}
+
 	resp, err = core.AlterCollection(ctx, &milvuspb.AlterCollectionRequest{
 		DbName:         dbName,
 		CollectionName: collectionName,
@@ -187,6 +215,46 @@ func TestDDLCallbacksAlterCollectionProperties(t *testing.T) {
 	require.ErrorIs(t, merr.CheckRPCCall(resp, err), merr.ErrParameterInvalid)
 }
 
+func TestValidateNamespaceModeImmutable(t *testing.T) {
+	for _, tc := range []struct {
+		name       string
+		properties []*commonpb.KeyValuePair
+		deleteKeys []string
+	}{
+		{
+			name:       "set namespace mode",
+			properties: []*commonpb.KeyValuePair{{Key: common.NamespaceModeKey, Value: common.NamespaceModePartition}},
+		},
+		{
+			name:       "set namespace mode with wrong case",
+			properties: []*commonpb.KeyValuePair{{Key: "Namespace.Mode", Value: common.NamespaceModePartition}},
+		},
+		{
+			name:       "delete namespace mode",
+			deleteKeys: []string{common.NamespaceModeKey},
+		},
+		{
+			name:       "delete namespace mode with wrong case",
+			deleteKeys: []string{"Namespace.Mode"},
+		},
+	} {
+		t.Run("reject "+tc.name, func(t *testing.T) {
+			err := validateNamespaceModeImmutable(tc.properties, tc.deleteKeys)
+			require.ErrorIs(t, err, merr.ErrParameterInvalid)
+			require.ErrorContains(t, err, common.NamespaceModeKey)
+		})
+	}
+
+	require.NoError(t, validateNamespaceModeImmutable(
+		[]*commonpb.KeyValuePair{{Key: common.CollectionReplicaNumber, Value: "2"}},
+		nil,
+	))
+	require.NoError(t, validateNamespaceModeImmutable(
+		nil,
+		[]string{common.CollectionReplicaNumber},
+	))
+}
+
 func TestDDLCallbacksAlterCollectionV2AckCallback_UpdateLoadConfigRPCError(t *testing.T) {
 	ctx := context.Background()
 

pkg/common/common.go

@@ -689,22 +689,16 @@ func IsNamespaceModePartition(kvs ...*commonpb.KeyValuePair) bool {
 	return GetNamespaceMode(kvs...) == NamespaceModePartition
 }
 
-type namespaceModeError string
-
-func (e namespaceModeError) Error() string {
-	return string(e)
-}
-
 func ValidateNamespaceMode(kvs ...*commonpb.KeyValuePair) error {
 	for _, kv := range kvs {
 		if kv.GetKey() == NamespaceModeKey {
 			if _, ok := normalizeNamespaceMode(kv.GetValue()); !ok {
-				return namespaceModeError("invalid namespace.mode value " + strconv.Quote(kv.GetValue()) + ", valid values: [" + ValidNamespaceModes + "]")
+				return merr.WrapErrParameterInvalidMsg("invalid namespace.mode value %q, valid values: [%s]", kv.GetValue(), ValidNamespaceModes)
 			}
 			return nil
 		}
 		if strings.EqualFold(kv.GetKey(), NamespaceModeKey) {
-			return namespaceModeError("invalid property key " + strconv.Quote(kv.GetKey()) + ", did you mean " + strconv.Quote(NamespaceModeKey) + "?")
+			return merr.WrapErrParameterInvalidMsg("invalid property key %q, did you mean %q?", kv.GetKey(), NamespaceModeKey)
 		}
 	}
 	return nil

pkg/common/common_test.go

@@ -8,6 +8,7 @@ import (
 	"github.com/stretchr/testify/assert"
 
 	"github.com/milvus-io/milvus-proto/go-api/v3/commonpb"
+	"github.com/milvus-io/milvus/pkg/v3/util/merr"
 )
 
 func TestIsSystemField(t *testing.T) {
@@ -186,6 +187,7 @@ func TestNamespaceMode(t *testing.T) {
 			}
 			err := ValidateNamespaceMode(kvs...)
 			assert.Error(t, err)
+			assert.ErrorIs(t, err, merr.ErrParameterInvalid)
 			assert.Contains(t, err.Error(), "valid values")
 		}
 	})
@@ -197,6 +199,7 @@ func TestNamespaceMode(t *testing.T) {
 			}
 			err := ValidateNamespaceMode(kvs...)
 			assert.Error(t, err, "value %q should be rejected", val)
+			assert.ErrorIs(t, err, merr.ErrParameterInvalid)
 			assert.Contains(t, err.Error(), "valid values")
 		}
 	})
@@ -208,6 +211,7 @@ func TestNamespaceMode(t *testing.T) {
 			}
 			err := ValidateNamespaceMode(kvs...)
 			assert.Error(t, err, "key %q should be rejected", key)
+			assert.ErrorIs(t, err, merr.ErrParameterInvalid)
 			assert.Contains(t, err.Error(), "did you mean")
 		}
 	})