enhance: support partition namespace creation (#50594)

## Summary
- Skip injecting `$namespace_id` when `namespace.mode=partition`.
- Avoid appending `partitionkey.isolation=true` for partition namespace
mode.
- Validate namespace mode before namespace field handling.

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

internal/rootcoord/create_collection_task.go

@@ -383,6 +383,13 @@ func (t *createCollectionTask) handleNamespaceField(ctx context.Context, schema
 		return nil
 	}
 
+	if common.IsNamespaceModePartition(t.Req.GetProperties()...) {
+		if hasPartitionKey {
+			return merr.WrapErrParameterInvalidMsg("namespace is not supported with partition key mode")
+		}
+		return nil
+	}
+
 	if typeutil.IsExternalCollection(schema) {
 		return merr.WrapErrParameterInvalidMsg("external collection does not support namespace field")
 	}
@@ -485,6 +492,9 @@ func (t *createCollectionTask) prepareSchema(ctx context.Context) error {
 		return err
 	}
 	t.appendDynamicField(ctx, t.body.CollectionSchema)
+	if err := common.ValidateNamespaceMode(t.Req.GetProperties()...); err != nil {
+		return err
+	}
 	if err := t.handleNamespaceField(ctx, t.body.CollectionSchema); err != nil {
 		return err
 	}
@@ -515,9 +525,6 @@ func (t *createCollectionTask) prepareSchema(ctx context.Context) error {
 	if exist && !timestamptz.IsTimezoneValid(tz) {
 		return merr.WrapErrParameterInvalidMsg("unknown or invalid IANA Time Zone ID: %s", tz)
 	}
-	if err := common.ValidateNamespaceMode(t.Req.GetProperties()...); err != nil {
-		return err
-	}
 
 	// Set properties for persistent
 	t.body.CollectionSchema.Properties = updateMaxFieldIDProperty(t.Req.GetProperties(), maxAssignedFieldIDFromSchema(t.body.CollectionSchema))

internal/rootcoord/create_collection_task_test.go

@@ -1894,7 +1894,8 @@ func TestCreateCollectionTask_Prepare_WithProperty(t *testing.T) {
 		core := newTestCore(withValidIDAllocator(), withTtSynchronizer(ticker), withMeta(meta))
 
 		schema := &schemapb.CollectionSchema{
-			Name: collectionName,
+			Name:            collectionName,
+			EnableNamespace: true,
 			Fields: []*schemapb.FieldSchema{
 				{Name: field1, DataType: schemapb.DataType_Int64},
 			},
@@ -1920,6 +1921,10 @@ func TestCreateCollectionTask_Prepare_WithProperty(t *testing.T) {
 		require.NoError(t, err)
 		props := common.CloneKeyValuePairs(task.body.CollectionSchema.Properties).ToMap()
 		assert.Equal(t, common.NamespaceModePartition, props[common.NamespaceModeKey])
+		assert.NotContains(t, props, common.PartitionKeyIsolationKey)
+		for _, field := range task.body.CollectionSchema.GetFields() {
+			assert.NotEqual(t, common.NamespaceFieldName, field.GetName())
+		}
 	})
 }
 
@@ -2142,6 +2147,27 @@ func TestNamespaceProperty(t *testing.T) {
 		assert.Error(t, err)
 	})
 
+	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 enabled with external collection", func(t *testing.T) {
 		// External collection is identified by having ExternalField set on fields
 		schema := &schemapb.CollectionSchema{