mirror of
https://github.com/milvus-io/milvus.git
synced 2026-07-21 10:15:43 +00:00
5def5ced4a
## What Enforce a function–field binding principle for function DDL (add / drop / alter function on an existing collection), so a function and its output field stay coupled and already-stored vectors can never be silently invalidated by DDL. BM25 and MinHash follow the strict binding: add/drop only via the unified `add_function_field` / `drop_function_field`, output field always coupled. **TextEmbedding is a temporary carve-out** — its legacy `AddCollectionFunction` / `DropCollectionFunction` paths are retained, so attach-over-existing-field and detach are still possible for it, because `add_function_field` embedding backfill is not yet implemented. It will be folded into the unified path in a follow-up. So the "always coupled" invariant currently holds for BM25/MinHash, not TextEmbedding. ## Changes - **AlterCollectionSchema invariant**: reject standalone add-function (a function must be added together with its new output field), reject detaching a function without dropping its output field, and reject function cascade (a new function's input being another function's output). - **Legacy RPCs**: `AddCollectionFunction` / `DropCollectionFunction` are rejected for BM25/MinHash (must use `add_function_field` / `drop_function_field`) and retained only for TextEmbedding (see carve-out). `AlterCollectionFunction` is kept and gains the whitelist below. In the alter and legacy-add paths, function field IDs are always re-derived from field names (never trusted from the request), so a request cannot inject an unrelated field ID that a later `drop_function_field` would delete. - **alter_function whitelist**: only connection/runtime params may change (TextEmbedding: `url`, `credential`, `timeout_ms`, `max_client_batch_size`, `region`, `location`, `projectid`, `user`); BM25/MinHash have no alterable params. Function identity (type, name, input/output fields) and output-shaping params (`dim`, `model_name`, `endpoint` — the TEI model identity, `normalize`, `truncate*`, prompts, ...) are immutable. Params are normalized (keys lowercased, duplicate keys rejected) so a crafted duplicate key cannot bypass the diff. - **drop_function_field**: allow any output-producing function (dropping needs no backfill), removing the previous BM25/MinHash-only restriction. ## Not in scope / follow-up - Extending `add_function_field` to TextEmbedding (post-creation add would require backfilling every existing row through the external embedding model) — deferred; folding the TextEmbedding legacy paths into the unified binding follows this. - MinHash input-field analyzer immutability lives on a different DDL path (`AlterCollectionField`); tracked as a separate follow-up. ## Breaking changes - `DropCollectionFunction` on a **MinHash** function (detach — remove the function but keep its output field) is no longer supported; use `drop_function_field` instead (drops the function together with its output field). If a released version supported MinHash detach, migrate any such call to `drop_function_field`. issue: #51348 🤖 Generated with [Claude Code](https://claude.com/claude-code) Signed-off-by: MrPresent-Han <chun.han@gmail.com> Co-authored-by: MrPresent-Han <chun.han@gmail.com> Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
1693 lines
69 KiB
Go
1693 lines
69 KiB
Go
// 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"
|
|
"strconv"
|
|
"strings"
|
|
"testing"
|
|
|
|
"github.com/cockroachdb/errors"
|
|
"github.com/stretchr/testify/mock"
|
|
"github.com/stretchr/testify/require"
|
|
"google.golang.org/protobuf/proto"
|
|
"google.golang.org/protobuf/types/known/fieldmaskpb"
|
|
|
|
"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"
|
|
"github.com/milvus-io/milvus/internal/metastore/model"
|
|
imocks "github.com/milvus-io/milvus/internal/mocks"
|
|
"github.com/milvus-io/milvus/internal/mocks/distributed/mock_streaming"
|
|
"github.com/milvus-io/milvus/internal/mocks/streamingcoord/server/mock_balancer"
|
|
"github.com/milvus-io/milvus/internal/streamingcoord/server/balancer/balance"
|
|
"github.com/milvus-io/milvus/internal/streamingcoord/server/broadcaster/registry"
|
|
"github.com/milvus-io/milvus/internal/tso"
|
|
mocktso "github.com/milvus-io/milvus/internal/tso/mocks"
|
|
"github.com/milvus-io/milvus/internal/util/schemautil"
|
|
"github.com/milvus-io/milvus/pkg/v3/common"
|
|
"github.com/milvus-io/milvus/pkg/v3/proto/indexpb"
|
|
"github.com/milvus-io/milvus/pkg/v3/proto/internalpb"
|
|
"github.com/milvus-io/milvus/pkg/v3/proto/messagespb"
|
|
"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/util/funcutil"
|
|
"github.com/milvus-io/milvus/pkg/v3/util/merr"
|
|
"github.com/milvus-io/milvus/pkg/v3/util/paramtable"
|
|
"github.com/milvus-io/milvus/pkg/v3/util/timestamptz"
|
|
"github.com/milvus-io/milvus/pkg/v3/util/typeutil"
|
|
)
|
|
|
|
// buildAlterSchemaReq constructs a valid AlterCollectionSchemaRequest with an add operation.
|
|
func buildAlterSchemaReq(dbName, collName, inputField, outputField, funcName string) *milvuspb.AlterCollectionSchemaRequest {
|
|
outputFieldSchema := &schemapb.FieldSchema{
|
|
Name: outputField,
|
|
DataType: schemapb.DataType_SparseFloatVector,
|
|
IsFunctionOutput: true,
|
|
}
|
|
functionSchema := &schemapb.FunctionSchema{
|
|
Name: funcName,
|
|
Type: schemapb.FunctionType_BM25,
|
|
InputFieldNames: []string{inputField},
|
|
OutputFieldNames: []string{outputField},
|
|
}
|
|
return &milvuspb.AlterCollectionSchemaRequest{
|
|
DbName: dbName,
|
|
CollectionName: collName,
|
|
Action: &milvuspb.AlterCollectionSchemaRequest_Action{
|
|
Op: &milvuspb.AlterCollectionSchemaRequest_Action_AddRequest{
|
|
AddRequest: &milvuspb.AlterCollectionSchemaRequest_AddRequest{
|
|
FieldInfos: []*milvuspb.AlterCollectionSchemaRequest_FieldInfo{
|
|
{
|
|
FieldSchema: outputFieldSchema,
|
|
ExtraParams: []*commonpb.KeyValuePair{
|
|
{Key: common.IndexTypeKey, Value: "SPARSE_INVERTED_INDEX"},
|
|
{Key: common.MetricTypeKey, Value: "BM25"},
|
|
},
|
|
},
|
|
},
|
|
FuncSchema: []*schemapb.FunctionSchema{functionSchema},
|
|
},
|
|
},
|
|
},
|
|
}
|
|
}
|
|
|
|
func buildAlterSchemaAddFieldReq(dbName, collName, fieldName string, doBackfill bool) *milvuspb.AlterCollectionSchemaRequest {
|
|
return buildAlterSchemaAddFieldSchemaReq(dbName, collName, &schemapb.FieldSchema{
|
|
Name: fieldName,
|
|
DataType: schemapb.DataType_VarChar,
|
|
Nullable: true,
|
|
TypeParams: []*commonpb.KeyValuePair{
|
|
{Key: common.MaxLengthKey, Value: "128"},
|
|
},
|
|
}, doBackfill)
|
|
}
|
|
|
|
func buildAlterSchemaAddFieldSchemaReq(dbName, collName string, fieldSchema *schemapb.FieldSchema, doBackfill bool) *milvuspb.AlterCollectionSchemaRequest {
|
|
return &milvuspb.AlterCollectionSchemaRequest{
|
|
DbName: dbName,
|
|
CollectionName: collName,
|
|
Action: &milvuspb.AlterCollectionSchemaRequest_Action{
|
|
Op: &milvuspb.AlterCollectionSchemaRequest_Action_AddRequest{
|
|
AddRequest: &milvuspb.AlterCollectionSchemaRequest_AddRequest{
|
|
FieldInfos: []*milvuspb.AlterCollectionSchemaRequest_FieldInfo{
|
|
{FieldSchema: fieldSchema},
|
|
},
|
|
DoPhysicalBackfill: doBackfill,
|
|
},
|
|
},
|
|
},
|
|
}
|
|
}
|
|
|
|
func buildAlterSchemaAddFunctionReq(dbName, collName string, functionSchema *schemapb.FunctionSchema) *milvuspb.AlterCollectionSchemaRequest {
|
|
return &milvuspb.AlterCollectionSchemaRequest{
|
|
DbName: dbName,
|
|
CollectionName: collName,
|
|
Action: &milvuspb.AlterCollectionSchemaRequest_Action{
|
|
Op: &milvuspb.AlterCollectionSchemaRequest_Action_AddRequest{
|
|
AddRequest: &milvuspb.AlterCollectionSchemaRequest_AddRequest{
|
|
FuncSchema: []*schemapb.FunctionSchema{functionSchema},
|
|
},
|
|
},
|
|
},
|
|
}
|
|
}
|
|
|
|
func TestDDLCallbacksBroadcastAlterCollectionSchema(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)
|
|
assertSchemaVersion(t, ctx, core, dbName, collectionName, 0)
|
|
|
|
// case 1: action == nil
|
|
resp, err := core.AlterCollectionSchema(ctx, &milvuspb.AlterCollectionSchemaRequest{
|
|
DbName: dbName,
|
|
CollectionName: collectionName,
|
|
Action: nil,
|
|
})
|
|
require.ErrorIs(t, merr.CheckRPCCall(resp.GetAlterStatus(), err), merr.ErrParameterInvalid)
|
|
|
|
// case 2: add_request == nil (action present but no Op set)
|
|
resp, err = core.AlterCollectionSchema(ctx, &milvuspb.AlterCollectionSchemaRequest{
|
|
DbName: dbName,
|
|
CollectionName: collectionName,
|
|
Action: &milvuspb.AlterCollectionSchemaRequest_Action{},
|
|
})
|
|
require.ErrorIs(t, merr.CheckRPCCall(resp.GetAlterStatus(), err), merr.ErrParameterInvalid)
|
|
|
|
// case 3: add a plain field without funcSchema.
|
|
resp, err = core.AlterCollectionSchema(ctx, buildAlterSchemaAddFieldReq(dbName, collectionName, "plain_text", false))
|
|
require.NoError(t, merr.CheckRPCCall(resp.GetAlterStatus(), err))
|
|
assertSchemaVersion(t, ctx, core, dbName, collectionName, 1)
|
|
coll, err := core.meta.GetCollectionByName(ctx, dbName, collectionName, typeutil.MaxTimestamp, false)
|
|
require.NoError(t, err)
|
|
require.Len(t, coll.Functions, 0)
|
|
plainFieldFound := false
|
|
for _, field := range coll.Fields {
|
|
if field.Name == "plain_text" {
|
|
plainFieldFound = true
|
|
require.False(t, field.IsFunctionOutput)
|
|
}
|
|
}
|
|
require.True(t, plainFieldFound)
|
|
|
|
// case 3.1: DoPhysicalBackfill is ignored by alter schema.
|
|
resp, err = core.AlterCollectionSchema(ctx, buildAlterSchemaAddFieldReq(dbName, collectionName, "plain_text_backfill", true))
|
|
require.NoError(t, merr.CheckRPCCall(resp.GetAlterStatus(), err))
|
|
assertSchemaVersion(t, ctx, core, dbName, collectionName, 2)
|
|
coll, err = core.meta.GetCollectionByName(ctx, dbName, collectionName, typeutil.MaxTimestamp, false)
|
|
require.NoError(t, err)
|
|
require.False(t, coll.ToCollectionSchemaPB().GetDoPhysicalBackfill())
|
|
|
|
status, err := core.AlterCollection(ctx, &milvuspb.AlterCollectionRequest{
|
|
DbName: dbName,
|
|
CollectionName: collectionName,
|
|
Properties: []*commonpb.KeyValuePair{
|
|
{Key: common.TimezoneKey, Value: "Asia/Shanghai"},
|
|
},
|
|
})
|
|
require.NoError(t, merr.CheckRPCCall(status, err))
|
|
|
|
// case 3.2: field-only TIMESTAMPTZ add rewrites string default value with collection timezone.
|
|
defaultTimeString := "2024-01-02T03:04:05"
|
|
resp, err = core.AlterCollectionSchema(ctx, buildAlterSchemaAddFieldSchemaReq(dbName, collectionName, &schemapb.FieldSchema{
|
|
Name: "created_at_tz",
|
|
DataType: schemapb.DataType_Timestamptz,
|
|
Nullable: true,
|
|
DefaultValue: &schemapb.ValueField{
|
|
Data: &schemapb.ValueField_StringData{StringData: defaultTimeString},
|
|
},
|
|
}, false))
|
|
require.NoError(t, merr.CheckRPCCall(resp.GetAlterStatus(), err))
|
|
assertSchemaVersion(t, ctx, core, dbName, collectionName, 3)
|
|
coll, err = core.meta.GetCollectionByName(ctx, dbName, collectionName, typeutil.MaxTimestamp, false)
|
|
require.NoError(t, err)
|
|
expectedTimestamptzDefault, err := timestamptz.ValidateAndReturnUnixMicroTz(defaultTimeString, "Asia/Shanghai")
|
|
require.NoError(t, err)
|
|
timestamptzDefaultFound := false
|
|
for _, field := range coll.Fields {
|
|
if field.Name == "created_at_tz" {
|
|
timestamptzDefaultFound = true
|
|
require.Equal(t, expectedTimestamptzDefault, field.DefaultValue.GetTimestamptzData())
|
|
require.Empty(t, field.DefaultValue.GetStringData())
|
|
}
|
|
}
|
|
require.True(t, timestamptzDefaultFound)
|
|
|
|
// case 3.3: field-only TIMESTAMPTZ add rejects invalid string default value.
|
|
resp, err = core.AlterCollectionSchema(ctx, buildAlterSchemaAddFieldSchemaReq(dbName, collectionName, &schemapb.FieldSchema{
|
|
Name: "invalid_created_at_tz",
|
|
DataType: schemapb.DataType_Timestamptz,
|
|
Nullable: true,
|
|
DefaultValue: &schemapb.ValueField{
|
|
Data: &schemapb.ValueField_StringData{StringData: "not-a-timestamp"},
|
|
},
|
|
}, false))
|
|
require.ErrorIs(t, merr.CheckRPCCall(resp.GetAlterStatus(), err), merr.ErrParameterInvalid)
|
|
|
|
// case 3.4: multiple function schemas remain unsupported.
|
|
resp, err = core.AlterCollectionSchema(ctx, &milvuspb.AlterCollectionSchemaRequest{
|
|
DbName: dbName,
|
|
CollectionName: collectionName,
|
|
Action: &milvuspb.AlterCollectionSchemaRequest_Action{
|
|
Op: &milvuspb.AlterCollectionSchemaRequest_Action_AddRequest{
|
|
AddRequest: &milvuspb.AlterCollectionSchemaRequest_AddRequest{
|
|
FuncSchema: []*schemapb.FunctionSchema{
|
|
{Name: "fn_multi_1", Type: schemapb.FunctionType_BM25},
|
|
{Name: "fn_multi_2", Type: schemapb.FunctionType_BM25},
|
|
},
|
|
FieldInfos: []*milvuspb.AlterCollectionSchemaRequest_FieldInfo{
|
|
{FieldSchema: &schemapb.FieldSchema{Name: "sparse1", DataType: schemapb.DataType_SparseFloatVector}},
|
|
},
|
|
},
|
|
},
|
|
},
|
|
})
|
|
require.ErrorIs(t, merr.CheckRPCCall(resp.GetAlterStatus(), err), merr.ErrParameterInvalid)
|
|
|
|
// case 3.5: nil function schema is rejected.
|
|
resp, err = core.AlterCollectionSchema(ctx, buildAlterSchemaAddFunctionReq(dbName, collectionName, nil))
|
|
require.ErrorIs(t, merr.CheckRPCCall(resp.GetAlterStatus(), err), merr.ErrParameterInvalid)
|
|
require.Contains(t, resp.GetAlterStatus().GetReason(), "function schema is nil")
|
|
|
|
// case 3.6: multiple fieldInfos remain unsupported.
|
|
resp, err = core.AlterCollectionSchema(ctx, &milvuspb.AlterCollectionSchemaRequest{
|
|
DbName: dbName,
|
|
CollectionName: collectionName,
|
|
Action: &milvuspb.AlterCollectionSchemaRequest_Action{
|
|
Op: &milvuspb.AlterCollectionSchemaRequest_Action_AddRequest{
|
|
AddRequest: &milvuspb.AlterCollectionSchemaRequest_AddRequest{
|
|
FieldInfos: []*milvuspb.AlterCollectionSchemaRequest_FieldInfo{
|
|
{FieldSchema: &schemapb.FieldSchema{Name: "plain_text_multi_1", DataType: schemapb.DataType_Int64}},
|
|
{FieldSchema: &schemapb.FieldSchema{Name: "plain_text_multi_2", DataType: schemapb.DataType_Int64}},
|
|
},
|
|
},
|
|
},
|
|
},
|
|
})
|
|
require.ErrorIs(t, merr.CheckRPCCall(resp.GetAlterStatus(), err), merr.ErrParameterInvalid)
|
|
|
|
// case 3.7: invalid field schema is rejected by RootCoord schema checks.
|
|
resp, err = core.AlterCollectionSchema(ctx, &milvuspb.AlterCollectionSchemaRequest{
|
|
DbName: dbName,
|
|
CollectionName: collectionName,
|
|
Action: &milvuspb.AlterCollectionSchemaRequest_Action{
|
|
Op: &milvuspb.AlterCollectionSchemaRequest_Action_AddRequest{
|
|
AddRequest: &milvuspb.AlterCollectionSchemaRequest_AddRequest{
|
|
FieldInfos: []*milvuspb.AlterCollectionSchemaRequest_FieldInfo{
|
|
{FieldSchema: &schemapb.FieldSchema{
|
|
Name: "invalid_primary",
|
|
DataType: schemapb.DataType_Int64,
|
|
IsPrimaryKey: true,
|
|
Nullable: true,
|
|
}},
|
|
},
|
|
},
|
|
},
|
|
},
|
|
})
|
|
require.ErrorIs(t, merr.CheckRPCCall(resp.GetAlterStatus(), err), merr.ErrParameterInvalid)
|
|
|
|
// case 4: both fieldInfos and funcSchema are empty.
|
|
resp, err = core.AlterCollectionSchema(ctx, &milvuspb.AlterCollectionSchemaRequest{
|
|
DbName: dbName,
|
|
CollectionName: collectionName,
|
|
Action: &milvuspb.AlterCollectionSchemaRequest_Action{
|
|
Op: &milvuspb.AlterCollectionSchemaRequest_Action_AddRequest{
|
|
AddRequest: &milvuspb.AlterCollectionSchemaRequest_AddRequest{},
|
|
},
|
|
},
|
|
})
|
|
require.ErrorIs(t, merr.CheckRPCCall(resp.GetAlterStatus(), err), merr.ErrParameterInvalid)
|
|
|
|
// case 4.1: BM25 function must add its output field in the same request.
|
|
resp, err = core.AlterCollectionSchema(ctx, &milvuspb.AlterCollectionSchemaRequest{
|
|
DbName: dbName,
|
|
CollectionName: collectionName,
|
|
Action: &milvuspb.AlterCollectionSchemaRequest_Action{
|
|
Op: &milvuspb.AlterCollectionSchemaRequest_Action_AddRequest{
|
|
AddRequest: &milvuspb.AlterCollectionSchemaRequest_AddRequest{
|
|
FuncSchema: []*schemapb.FunctionSchema{
|
|
{Name: "fn1", Type: schemapb.FunctionType_BM25, InputFieldNames: []string{"field1"}, OutputFieldNames: []string{"sparse1"}},
|
|
},
|
|
FieldInfos: []*milvuspb.AlterCollectionSchemaRequest_FieldInfo{},
|
|
},
|
|
},
|
|
},
|
|
})
|
|
require.ErrorIs(t, merr.CheckRPCCall(resp.GetAlterStatus(), err), merr.ErrParameterInvalid)
|
|
require.Contains(t, resp.GetAlterStatus().GetReason(), "add_function_field")
|
|
|
|
// case 4.2: function-only output field must exist.
|
|
resp, err = core.AlterCollectionSchema(ctx, &milvuspb.AlterCollectionSchemaRequest{
|
|
DbName: dbName,
|
|
CollectionName: collectionName,
|
|
Action: &milvuspb.AlterCollectionSchemaRequest_Action{
|
|
Op: &milvuspb.AlterCollectionSchemaRequest_Action_AddRequest{
|
|
AddRequest: &milvuspb.AlterCollectionSchemaRequest_AddRequest{
|
|
FuncSchema: []*schemapb.FunctionSchema{
|
|
{
|
|
Name: "minhash_missing_output",
|
|
Type: schemapb.FunctionType_MinHash,
|
|
InputFieldNames: []string{"field1"},
|
|
OutputFieldNames: []string{"missing_minhash_output"},
|
|
Params: []*commonpb.KeyValuePair{
|
|
{Key: "num_hashes", Value: "128"},
|
|
{Key: "shingle_size", Value: "3"},
|
|
{Key: "hash_function", Value: "xxhash64"},
|
|
{Key: "seed", Value: "42"},
|
|
},
|
|
},
|
|
},
|
|
},
|
|
},
|
|
},
|
|
})
|
|
require.ErrorIs(t, merr.CheckRPCCall(resp.GetAlterStatus(), err), merr.ErrParameterInvalid)
|
|
require.Contains(t, resp.GetAlterStatus().GetReason(), "adding a function over existing fields is not supported")
|
|
|
|
// case 4.3: BM25 function with multiple output fields is rejected.
|
|
resp, err = core.AlterCollectionSchema(ctx, &milvuspb.AlterCollectionSchemaRequest{
|
|
DbName: dbName,
|
|
CollectionName: collectionName,
|
|
Action: &milvuspb.AlterCollectionSchemaRequest_Action{
|
|
Op: &milvuspb.AlterCollectionSchemaRequest_Action_AddRequest{
|
|
AddRequest: &milvuspb.AlterCollectionSchemaRequest_AddRequest{
|
|
FuncSchema: []*schemapb.FunctionSchema{
|
|
{
|
|
Name: "bm25_multi_output",
|
|
Type: schemapb.FunctionType_BM25,
|
|
InputFieldNames: []string{"field1"},
|
|
OutputFieldNames: []string{"sparse_multi_output", "sparse_multi_output_extra"},
|
|
},
|
|
},
|
|
FieldInfos: []*milvuspb.AlterCollectionSchemaRequest_FieldInfo{
|
|
{FieldSchema: &schemapb.FieldSchema{
|
|
Name: "sparse_multi_output",
|
|
DataType: schemapb.DataType_SparseFloatVector,
|
|
}},
|
|
},
|
|
},
|
|
},
|
|
},
|
|
})
|
|
require.ErrorIs(t, merr.CheckRPCCall(resp.GetAlterStatus(), err), merr.ErrParameterInvalid)
|
|
require.Contains(t, resp.GetAlterStatus().GetReason(), "exactly one output field")
|
|
|
|
// case 5: BM25 arity invalid
|
|
resp, err = core.AlterCollectionSchema(ctx, &milvuspb.AlterCollectionSchemaRequest{
|
|
DbName: dbName,
|
|
CollectionName: collectionName,
|
|
Action: &milvuspb.AlterCollectionSchemaRequest_Action{
|
|
Op: &milvuspb.AlterCollectionSchemaRequest_Action_AddRequest{
|
|
AddRequest: &milvuspb.AlterCollectionSchemaRequest_AddRequest{
|
|
FuncSchema: []*schemapb.FunctionSchema{
|
|
{
|
|
Name: "fn_bad_arity",
|
|
Type: schemapb.FunctionType_BM25,
|
|
InputFieldNames: []string{"field1", "field2", "field3"},
|
|
OutputFieldNames: []string{"sparse_bad_arity"},
|
|
},
|
|
},
|
|
FieldInfos: []*milvuspb.AlterCollectionSchemaRequest_FieldInfo{
|
|
{FieldSchema: &schemapb.FieldSchema{Name: "sparse_bad_arity", DataType: schemapb.DataType_SparseFloatVector}},
|
|
},
|
|
},
|
|
},
|
|
},
|
|
})
|
|
arityErr := merr.CheckRPCCall(resp.GetAlterStatus(), err)
|
|
require.ErrorIs(t, arityErr, merr.ErrParameterInvalid)
|
|
require.ErrorContains(t, arityErr, "exactly one input field and exactly one output field")
|
|
|
|
// case 6: fieldSchema nil in fieldInfos
|
|
resp, err = core.AlterCollectionSchema(ctx, &milvuspb.AlterCollectionSchemaRequest{
|
|
DbName: dbName,
|
|
CollectionName: collectionName,
|
|
Action: &milvuspb.AlterCollectionSchemaRequest_Action{
|
|
Op: &milvuspb.AlterCollectionSchemaRequest_Action_AddRequest{
|
|
AddRequest: &milvuspb.AlterCollectionSchemaRequest_AddRequest{
|
|
FuncSchema: []*schemapb.FunctionSchema{
|
|
{Name: "fn1", Type: schemapb.FunctionType_BM25, InputFieldNames: []string{"field1"}, OutputFieldNames: []string{"sparse_nil_field"}},
|
|
},
|
|
FieldInfos: []*milvuspb.AlterCollectionSchemaRequest_FieldInfo{
|
|
{FieldSchema: nil},
|
|
},
|
|
},
|
|
},
|
|
},
|
|
})
|
|
require.ErrorIs(t, merr.CheckRPCCall(resp.GetAlterStatus(), err), merr.ErrParameterInvalid)
|
|
|
|
// case 7: output field already exists (field1 created by createCollectionForTest)
|
|
resp, err = core.AlterCollectionSchema(ctx, &milvuspb.AlterCollectionSchemaRequest{
|
|
DbName: dbName,
|
|
CollectionName: collectionName,
|
|
Action: &milvuspb.AlterCollectionSchemaRequest_Action{
|
|
Op: &milvuspb.AlterCollectionSchemaRequest_Action_AddRequest{
|
|
AddRequest: &milvuspb.AlterCollectionSchemaRequest_AddRequest{
|
|
FuncSchema: []*schemapb.FunctionSchema{
|
|
{Name: "fn_dup_field", Type: schemapb.FunctionType_BM25, InputFieldNames: []string{"field1"}, OutputFieldNames: []string{"field1"}},
|
|
},
|
|
FieldInfos: []*milvuspb.AlterCollectionSchemaRequest_FieldInfo{
|
|
{
|
|
FieldSchema: &schemapb.FieldSchema{Name: "field1", DataType: schemapb.DataType_SparseFloatVector},
|
|
ExtraParams: []*commonpb.KeyValuePair{
|
|
{Key: common.IndexTypeKey, Value: "SPARSE_INVERTED_INDEX"},
|
|
{Key: common.MetricTypeKey, Value: "BM25"},
|
|
},
|
|
},
|
|
},
|
|
},
|
|
},
|
|
},
|
|
})
|
|
require.ErrorIs(t, merr.CheckRPCCall(resp.GetAlterStatus(), err), merr.ErrParameterInvalid)
|
|
|
|
// case 7.1: vector function output field without bound index params is rejected.
|
|
noIndexReq := buildAlterSchemaReq(dbName, collectionName, "field1", "sparse_no_index", "fn_no_index")
|
|
noIndexReq.GetAction().GetAddRequest().GetFieldInfos()[0].ExtraParams = nil
|
|
resp, err = core.AlterCollectionSchema(ctx, noIndexReq)
|
|
noIndexErr := merr.CheckRPCCall(resp.GetAlterStatus(), err)
|
|
require.ErrorIs(t, noIndexErr, merr.ErrParameterInvalid)
|
|
require.ErrorContains(t, noIndexErr, "index params are required")
|
|
|
|
// case 8: output field points to an existing field while FieldInfos adds a different field
|
|
resp, err = core.AlterCollectionSchema(ctx, &milvuspb.AlterCollectionSchemaRequest{
|
|
DbName: dbName,
|
|
CollectionName: collectionName,
|
|
Action: &milvuspb.AlterCollectionSchemaRequest_Action{
|
|
Op: &milvuspb.AlterCollectionSchemaRequest_Action_AddRequest{
|
|
AddRequest: &milvuspb.AlterCollectionSchemaRequest_AddRequest{
|
|
FuncSchema: []*schemapb.FunctionSchema{
|
|
{Name: "fn_output_existing_field", Type: schemapb.FunctionType_BM25, InputFieldNames: []string{"field1"}, OutputFieldNames: []string{"field1"}},
|
|
},
|
|
FieldInfos: []*milvuspb.AlterCollectionSchemaRequest_FieldInfo{
|
|
{FieldSchema: &schemapb.FieldSchema{Name: "sparse_output_existing_bypass", DataType: schemapb.DataType_SparseFloatVector, IsFunctionOutput: true}},
|
|
},
|
|
},
|
|
},
|
|
},
|
|
})
|
|
require.ErrorIs(t, merr.CheckRPCCall(resp.GetAlterStatus(), err), merr.ErrParameterInvalid)
|
|
|
|
// Add a VARCHAR input field so that the happy-path call below can succeed.
|
|
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"},
|
|
},
|
|
}
|
|
varcharBytes, err := proto.Marshal(varcharFieldSchema)
|
|
require.NoError(t, err)
|
|
addFieldResp, err := core.AddCollectionField(ctx, &milvuspb.AddCollectionFieldRequest{
|
|
DbName: dbName,
|
|
CollectionName: collectionName,
|
|
Schema: varcharBytes,
|
|
})
|
|
require.NoError(t, merr.CheckRPCCall(addFieldResp, err))
|
|
|
|
// case 7.2: AUTOINDEX is not accepted as the bound index type in V1
|
|
// (rejected at rootcoord prepare, after function validation passes).
|
|
autoIndexReq := buildAlterSchemaReq(dbName, collectionName, "text_input", "sparse_auto_index", "fn_auto_index")
|
|
autoIndexReq.GetAction().GetAddRequest().GetFieldInfos()[0].ExtraParams = []*commonpb.KeyValuePair{
|
|
{Key: common.IndexTypeKey, Value: common.AutoIndexName},
|
|
}
|
|
resp, err = core.AlterCollectionSchema(ctx, autoIndexReq)
|
|
autoIndexErr := merr.CheckRPCCall(resp.GetAlterStatus(), err)
|
|
require.ErrorIs(t, autoIndexErr, merr.ErrParameterInvalid)
|
|
require.ErrorContains(t, autoIndexErr, "explicit index_type is required")
|
|
|
|
// case 7.3: an index type without a registered checker is rejected at prepare —
|
|
// it would otherwise be persisted via the ack callback and never build.
|
|
unknownIndexReq := buildAlterSchemaReq(dbName, collectionName, "text_input", "sparse_unknown_index", "fn_unknown_index")
|
|
unknownIndexReq.GetAction().GetAddRequest().GetFieldInfos()[0].ExtraParams = []*commonpb.KeyValuePair{
|
|
{Key: common.IndexTypeKey, Value: "NOT_A_REAL_INDEX"},
|
|
{Key: common.MetricTypeKey, Value: "BM25"},
|
|
}
|
|
resp, err = core.AlterCollectionSchema(ctx, unknownIndexReq)
|
|
unknownIndexErr := merr.CheckRPCCall(resp.GetAlterStatus(), err)
|
|
require.ErrorIs(t, unknownIndexErr, merr.ErrParameterInvalid)
|
|
require.ErrorContains(t, unknownIndexErr, "invalid index type")
|
|
|
|
// happy path: add binary vector output field + MinHash function.
|
|
minHashReq := buildAlterSchemaReq(dbName, collectionName, "text_input", "binary_minhash_output", "minhash_fn")
|
|
minHashFieldSchema := minHashReq.GetAction().GetAddRequest().GetFieldInfos()[0].GetFieldSchema()
|
|
minHashFieldSchema.DataType = schemapb.DataType_BinaryVector
|
|
minHashFieldSchema.TypeParams = []*commonpb.KeyValuePair{
|
|
{Key: common.DimKey, Value: "4096"},
|
|
}
|
|
minHashReq.GetAction().GetAddRequest().GetFieldInfos()[0].ExtraParams = []*commonpb.KeyValuePair{
|
|
{Key: common.IndexTypeKey, Value: "MINHASH_LSH"},
|
|
{Key: common.MetricTypeKey, Value: "MHJACCARD"},
|
|
}
|
|
minHashFunction := minHashReq.GetAction().GetAddRequest().GetFuncSchema()[0]
|
|
minHashFunction.Type = schemapb.FunctionType_MinHash
|
|
minHashFunction.Params = []*commonpb.KeyValuePair{
|
|
{Key: "num_hashes", Value: "128"},
|
|
{Key: "shingle_size", Value: "3"},
|
|
{Key: "hash_function", Value: "xxhash64"},
|
|
{Key: "seed", Value: "42"},
|
|
}
|
|
resp, err = core.AlterCollectionSchema(ctx, minHashReq)
|
|
require.NoError(t, merr.CheckRPCCall(resp.GetAlterStatus(), err))
|
|
assertSchemaVersion(t, ctx, core, dbName, collectionName, 5)
|
|
|
|
// The bound index meta must have been applied through the CreateIndex ack
|
|
// callback within the same DDL, fully materialized at prepare time.
|
|
coll, err = core.meta.GetCollectionByName(ctx, dbName, collectionName, typeutil.MaxTimestamp, false)
|
|
require.NoError(t, err)
|
|
var minHashFieldID int64
|
|
for _, field := range coll.Fields {
|
|
if field.Name == "binary_minhash_output" {
|
|
minHashFieldID = field.FieldID
|
|
}
|
|
}
|
|
require.Positive(t, minHashFieldID)
|
|
boundIndexes := recordedBoundIndexes()
|
|
require.Len(t, boundIndexes, 1)
|
|
boundIndexInfo := boundIndexes[0].GetIndexInfo()
|
|
require.Equal(t, "binary_minhash_output", boundIndexInfo.GetIndexName())
|
|
require.Equal(t, minHashFieldID, boundIndexInfo.GetFieldID())
|
|
require.Positive(t, boundIndexInfo.GetIndexID())
|
|
boundIndexParams := funcutil.KeyValuePair2Map(boundIndexInfo.GetIndexParams())
|
|
require.Equal(t, "MINHASH_LSH", boundIndexParams[common.IndexTypeKey])
|
|
require.Equal(t, "MHJACCARD", boundIndexParams[common.MetricTypeKey])
|
|
|
|
minHashBadArityReq := buildAlterSchemaReq(dbName, collectionName, "text_input", "minhash_bad_arity", "minhash_bad_arity_fn")
|
|
minHashBadArityReq.GetAction().GetAddRequest().GetFuncSchema()[0].Type = schemapb.FunctionType_MinHash
|
|
minHashBadArityReq.GetAction().GetAddRequest().GetFuncSchema()[0].InputFieldNames = []string{"text_input", "field1"}
|
|
minHashBadArityReq.GetAction().GetAddRequest().GetFieldInfos()[0].GetFieldSchema().DataType = schemapb.DataType_BinaryVector
|
|
minHashBadArityReq.GetAction().GetAddRequest().GetFieldInfos()[0].GetFieldSchema().TypeParams = []*commonpb.KeyValuePair{
|
|
{Key: common.DimKey, Value: "4096"},
|
|
}
|
|
resp, err = core.AlterCollectionSchema(ctx, minHashBadArityReq)
|
|
alterErr := merr.CheckRPCCall(resp.GetAlterStatus(), err)
|
|
require.ErrorIs(t, alterErr, merr.ErrParameterInvalid)
|
|
require.ErrorContains(t, alterErr, "MinHash function should have exactly one input field and exactly one output field")
|
|
|
|
nullableOutputReq := buildAlterSchemaReq(dbName, collectionName, "text_input", "sparse_output_nullable", "bm25_nullable_output")
|
|
nullableOutputReq.GetAction().GetAddRequest().GetFieldInfos()[0].GetFieldSchema().Nullable = true
|
|
resp, err = core.AlterCollectionSchema(ctx, nullableOutputReq)
|
|
alterErr = merr.CheckRPCCall(resp.GetAlterStatus(), err)
|
|
require.ErrorIs(t, alterErr, merr.ErrParameterInvalid)
|
|
require.ErrorContains(t, alterErr, "function output field cannot be nullable")
|
|
|
|
existingOutputFieldSchema := &schemapb.FieldSchema{
|
|
Name: "existing_minhash_output",
|
|
DataType: schemapb.DataType_BinaryVector,
|
|
Nullable: true,
|
|
TypeParams: []*commonpb.KeyValuePair{
|
|
{Key: common.DimKey, Value: "4096"},
|
|
},
|
|
}
|
|
existingOutputBytes, err := proto.Marshal(existingOutputFieldSchema)
|
|
require.NoError(t, err)
|
|
addFieldResp, err = core.AddCollectionField(ctx, &milvuspb.AddCollectionFieldRequest{
|
|
DbName: dbName,
|
|
CollectionName: collectionName,
|
|
Schema: existingOutputBytes,
|
|
})
|
|
require.NoError(t, merr.CheckRPCCall(addFieldResp, err))
|
|
|
|
resp, err = core.AlterCollectionSchema(ctx, buildAlterSchemaAddFunctionReq(dbName, collectionName, &schemapb.FunctionSchema{
|
|
Name: "minhash_missing_output_late",
|
|
Type: schemapb.FunctionType_MinHash,
|
|
InputFieldNames: []string{"text_input"},
|
|
OutputFieldNames: []string{"missing_minhash_output_late"},
|
|
Params: []*commonpb.KeyValuePair{
|
|
{Key: "num_hashes", Value: "128"},
|
|
{Key: "shingle_size", Value: "3"},
|
|
{Key: "hash_function", Value: "xxhash64"},
|
|
{Key: "seed", Value: "42"},
|
|
},
|
|
}))
|
|
require.ErrorIs(t, merr.CheckRPCCall(resp.GetAlterStatus(), err), merr.ErrParameterInvalid)
|
|
require.Contains(t, resp.GetAlterStatus().GetReason(), "adding a function over existing fields is not supported")
|
|
|
|
// function-only add (marking an existing field as output) is no longer supported.
|
|
functionOnlyReq := buildAlterSchemaAddFunctionReq(dbName, collectionName, &schemapb.FunctionSchema{
|
|
Name: "minhash_existing_fn",
|
|
Type: schemapb.FunctionType_MinHash,
|
|
InputFieldNames: []string{"text_input"},
|
|
OutputFieldNames: []string{"existing_minhash_output"},
|
|
Params: []*commonpb.KeyValuePair{
|
|
{Key: "num_hashes", Value: "128"},
|
|
{Key: "shingle_size", Value: "3"},
|
|
{Key: "hash_function", Value: "xxhash64"},
|
|
{Key: "seed", Value: "42"},
|
|
},
|
|
})
|
|
resp, err = core.AlterCollectionSchema(ctx, functionOnlyReq)
|
|
functionOnlyErr := merr.CheckRPCCall(resp.GetAlterStatus(), err)
|
|
require.ErrorIs(t, functionOnlyErr, merr.ErrParameterInvalid)
|
|
require.ErrorContains(t, functionOnlyErr, "adding a function over existing fields is not supported")
|
|
|
|
// happy path: add sparse vector output field + BM25 function.
|
|
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, 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, 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, 8, schema.GetVersion())
|
|
|
|
// case 9: function already exists (same name "bm25_fn")
|
|
resp, err = core.AlterCollectionSchema(ctx, &milvuspb.AlterCollectionSchemaRequest{
|
|
DbName: dbName,
|
|
CollectionName: collectionName,
|
|
Action: &milvuspb.AlterCollectionSchemaRequest_Action{
|
|
Op: &milvuspb.AlterCollectionSchemaRequest_Action_AddRequest{
|
|
AddRequest: &milvuspb.AlterCollectionSchemaRequest_AddRequest{
|
|
FuncSchema: []*schemapb.FunctionSchema{
|
|
{Name: "bm25_fn", Type: schemapb.FunctionType_BM25, InputFieldNames: []string{"text_input"}, OutputFieldNames: []string{"sparse_output2"}},
|
|
},
|
|
FieldInfos: []*milvuspb.AlterCollectionSchemaRequest_FieldInfo{
|
|
{FieldSchema: &schemapb.FieldSchema{Name: "sparse_output2", DataType: schemapb.DataType_SparseFloatVector, IsFunctionOutput: true}},
|
|
},
|
|
},
|
|
},
|
|
},
|
|
})
|
|
require.ErrorIs(t, merr.CheckRPCCall(resp.GetAlterStatus(), err), merr.ErrParameterInvalid)
|
|
|
|
// case 10: input field not found
|
|
resp, err = core.AlterCollectionSchema(ctx, &milvuspb.AlterCollectionSchemaRequest{
|
|
DbName: dbName,
|
|
CollectionName: collectionName,
|
|
Action: &milvuspb.AlterCollectionSchemaRequest_Action{
|
|
Op: &milvuspb.AlterCollectionSchemaRequest_Action_AddRequest{
|
|
AddRequest: &milvuspb.AlterCollectionSchemaRequest_AddRequest{
|
|
FuncSchema: []*schemapb.FunctionSchema{
|
|
{Name: "fn_missing_input", Type: schemapb.FunctionType_BM25, InputFieldNames: []string{"nonexistent_input"}, OutputFieldNames: []string{"sparse_output3"}},
|
|
},
|
|
FieldInfos: []*milvuspb.AlterCollectionSchemaRequest_FieldInfo{
|
|
{FieldSchema: &schemapb.FieldSchema{Name: "sparse_output3", DataType: schemapb.DataType_SparseFloatVector, IsFunctionOutput: true}},
|
|
},
|
|
},
|
|
},
|
|
},
|
|
})
|
|
require.Error(t, merr.CheckRPCCall(resp.GetAlterStatus(), err))
|
|
|
|
// case 11: output field not found (function references "ghost_output" but FieldInfos has "sparse_output4")
|
|
resp, err = core.AlterCollectionSchema(ctx, &milvuspb.AlterCollectionSchemaRequest{
|
|
DbName: dbName,
|
|
CollectionName: collectionName,
|
|
Action: &milvuspb.AlterCollectionSchemaRequest_Action{
|
|
Op: &milvuspb.AlterCollectionSchemaRequest_Action_AddRequest{
|
|
AddRequest: &milvuspb.AlterCollectionSchemaRequest_AddRequest{
|
|
FuncSchema: []*schemapb.FunctionSchema{
|
|
{Name: "fn_missing_output", Type: schemapb.FunctionType_BM25, InputFieldNames: []string{"text_input"}, OutputFieldNames: []string{"ghost_output"}},
|
|
},
|
|
FieldInfos: []*milvuspb.AlterCollectionSchemaRequest_FieldInfo{
|
|
{FieldSchema: &schemapb.FieldSchema{Name: "sparse_output4", DataType: schemapb.DataType_SparseFloatVector, IsFunctionOutput: true}},
|
|
},
|
|
},
|
|
},
|
|
},
|
|
})
|
|
require.Error(t, merr.CheckRPCCall(resp.GetAlterStatus(), err))
|
|
}
|
|
|
|
func TestDDLCallbacksAlterCollectionSchemaAnalyzerFileResourceRefs(t *testing.T) {
|
|
core := initStreamingSystemAndCore(t)
|
|
|
|
ctx := context.Background()
|
|
dbName := "testDB" + funcutil.RandomString(10)
|
|
collectionName := "testCollection" + funcutil.RandomString(10)
|
|
fieldName := "schema_text_with_dict"
|
|
|
|
createCollectionForTest(t, ctx, core, dbName, collectionName)
|
|
|
|
meta := core.meta.(*MetaTable)
|
|
resourceID := int64(10002)
|
|
meta.fileResourceName2Meta = map[string]*internalpb.FileResourceInfo{
|
|
"schema_dict": {Id: resourceID, Name: "schema_dict", Path: "schema_dict.txt"},
|
|
}
|
|
meta.fileResourceID2Meta = map[int64]*internalpb.FileResourceInfo{
|
|
resourceID: {Id: resourceID, Name: "schema_dict", Path: "schema_dict.txt"},
|
|
}
|
|
meta.fileResourceRefCnt = map[int64]int{}
|
|
|
|
mixCoord := core.mixCoord.(*imocks.MixCoord)
|
|
mixCoord.EXPECT().ValidateAnalyzer(mock.Anything, mock.MatchedBy(func(req *querypb.ValidateAnalyzerRequest) bool {
|
|
infos := req.GetAnalyzerInfos()
|
|
return len(infos) == 1 &&
|
|
infos[0].GetField() == fieldName &&
|
|
infos[0].GetParams() == `{"tokenizer":"standard"}`
|
|
})).Return(&querypb.ValidateAnalyzerResponse{
|
|
Status: merr.Success(),
|
|
ResourceIds: []int64{resourceID},
|
|
}, nil).Once()
|
|
|
|
resp, err := core.AlterCollectionSchema(ctx, buildAlterSchemaAddFieldSchemaReq(dbName, collectionName, &schemapb.FieldSchema{
|
|
Name: fieldName,
|
|
DataType: schemapb.DataType_VarChar,
|
|
Nullable: true,
|
|
TypeParams: []*commonpb.KeyValuePair{
|
|
{Key: common.MaxLengthKey, Value: "128"},
|
|
{Key: common.EnableAnalyzerKey, Value: "true"},
|
|
{Key: common.AnalyzerParamKey, Value: `{"tokenizer":"standard"}`},
|
|
},
|
|
}, false))
|
|
require.NoError(t, merr.CheckRPCCall(resp.GetAlterStatus(), err))
|
|
coll, err := core.meta.GetCollectionByName(ctx, dbName, collectionName, typeutil.MaxTimestamp, false)
|
|
require.NoError(t, err)
|
|
require.ElementsMatch(t, []int64{resourceID}, coll.FileResourceIds)
|
|
require.Equal(t, 1, meta.fileResourceRefCnt[resourceID])
|
|
|
|
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: fieldName},
|
|
},
|
|
},
|
|
},
|
|
})
|
|
require.NoError(t, merr.CheckRPCCall(resp.GetAlterStatus(), err))
|
|
coll, err = core.meta.GetCollectionByName(ctx, dbName, collectionName, typeutil.MaxTimestamp, false)
|
|
require.NoError(t, err)
|
|
require.Empty(t, coll.FileResourceIds)
|
|
require.Equal(t, 0, meta.fileResourceRefCnt[resourceID])
|
|
assertFieldNotExists(t, ctx, core, dbName, collectionName, fieldName)
|
|
}
|
|
|
|
func TestDDLCallbacksAlterCollectionSchemaRejectsFunctionOnlyAdd(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)
|
|
|
|
inputFieldBytes, err := proto.Marshal(&schemapb.FieldSchema{
|
|
Name: "text_input",
|
|
DataType: schemapb.DataType_VarChar,
|
|
Nullable: true,
|
|
TypeParams: []*commonpb.KeyValuePair{
|
|
{Key: common.MaxLengthKey, Value: "256"},
|
|
},
|
|
})
|
|
require.NoError(t, err)
|
|
addFieldResp, err := core.AddCollectionField(ctx, &milvuspb.AddCollectionFieldRequest{
|
|
DbName: dbName,
|
|
CollectionName: collectionName,
|
|
Schema: inputFieldBytes,
|
|
})
|
|
require.NoError(t, merr.CheckRPCCall(addFieldResp, err))
|
|
|
|
invalidOutputBytes, err := proto.Marshal(&schemapb.FieldSchema{
|
|
Name: "invalid_minhash_output",
|
|
DataType: schemapb.DataType_FloatVector,
|
|
Nullable: true,
|
|
TypeParams: []*commonpb.KeyValuePair{
|
|
{Key: common.DimKey, Value: "128"},
|
|
},
|
|
})
|
|
require.NoError(t, err)
|
|
addFieldResp, err = core.AddCollectionField(ctx, &milvuspb.AddCollectionFieldRequest{
|
|
DbName: dbName,
|
|
CollectionName: collectionName,
|
|
Schema: invalidOutputBytes,
|
|
})
|
|
require.NoError(t, merr.CheckRPCCall(addFieldResp, err))
|
|
assertSchemaVersion(t, ctx, core, dbName, collectionName, 2)
|
|
|
|
resp, err := core.AlterCollectionSchema(ctx, buildAlterSchemaAddFunctionReq(dbName, collectionName, &schemapb.FunctionSchema{
|
|
Name: "minhash_invalid_output",
|
|
Type: schemapb.FunctionType_MinHash,
|
|
InputFieldNames: []string{"text_input"},
|
|
OutputFieldNames: []string{"invalid_minhash_output"},
|
|
Params: []*commonpb.KeyValuePair{
|
|
{Key: "num_hashes", Value: "128"},
|
|
{Key: "shingle_size", Value: "3"},
|
|
{Key: "hash_function", Value: "xxhash64"},
|
|
{Key: "seed", Value: "42"},
|
|
},
|
|
}))
|
|
alterErr := merr.CheckRPCCall(resp.GetAlterStatus(), err)
|
|
require.ErrorIs(t, alterErr, merr.ErrParameterInvalid)
|
|
require.ErrorContains(t, alterErr, "adding a function over existing fields is not supported")
|
|
assertSchemaVersion(t, ctx, core, dbName, collectionName, 2)
|
|
}
|
|
|
|
func TestDDLCallbacksAlterCollectionSchemaAddRejectsStructFieldNameConflicts(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.AddCollectionStructField(ctx, &milvuspb.AddCollectionStructFieldRequest{
|
|
DbName: dbName,
|
|
CollectionName: collectionName,
|
|
StructArrayFieldSchema: newRootAddStructFieldSchema("profile"),
|
|
})
|
|
require.NoError(t, merr.CheckRPCCall(resp, err))
|
|
assertSchemaVersion(t, ctx, core, dbName, collectionName, 1)
|
|
|
|
for _, fieldName := range []string{"profile", "profile[ints]", storedRootStructSubFieldName("profile", "profile[ints]")} {
|
|
t.Run(fieldName, func(t *testing.T) {
|
|
resp, err := core.AlterCollectionSchema(ctx, buildAlterSchemaAddFieldReq(dbName, collectionName, fieldName, false))
|
|
require.ErrorIs(t, merr.CheckRPCCall(resp.GetAlterStatus(), err), merr.ErrParameterInvalid)
|
|
require.Contains(t, resp.GetAlterStatus().GetReason(), "field already exists")
|
|
assertSchemaVersion(t, ctx, core, dbName, collectionName, 1)
|
|
})
|
|
}
|
|
}
|
|
|
|
func TestDDLCallbacksAlterCollectionDropField(t *testing.T) {
|
|
core := initStreamingSystemAndCore(t)
|
|
|
|
ctx := context.Background()
|
|
dbName := "testDB" + funcutil.RandomString(10)
|
|
collectionName := "testCollection" + funcutil.RandomString(10)
|
|
|
|
// database not found
|
|
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: "field1"},
|
|
},
|
|
},
|
|
},
|
|
})
|
|
require.ErrorIs(t, merr.CheckRPCCall(resp.GetAlterStatus(), err), merr.ErrDatabaseNotFound)
|
|
|
|
// create collection with field1
|
|
createCollectionForTest(t, ctx, core, dbName, collectionName)
|
|
|
|
// add field2 and field3
|
|
addResp, err := core.AddCollectionField(ctx, &milvuspb.AddCollectionFieldRequest{
|
|
DbName: dbName,
|
|
CollectionName: collectionName,
|
|
Schema: getFieldSchema("field2"),
|
|
})
|
|
require.NoError(t, merr.CheckRPCCall(addResp, err))
|
|
assertFieldExists(t, ctx, core, dbName, collectionName, "field2", 101)
|
|
|
|
addResp, err = core.AddCollectionField(ctx, &milvuspb.AddCollectionFieldRequest{
|
|
DbName: dbName,
|
|
CollectionName: collectionName,
|
|
Schema: getFieldSchema("field3"),
|
|
})
|
|
require.NoError(t, merr.CheckRPCCall(addResp, err))
|
|
assertFieldExists(t, ctx, core, dbName, collectionName, "field3", 102)
|
|
assertSchemaVersion(t, ctx, core, dbName, collectionName, 2)
|
|
|
|
// helper to build drop-field request
|
|
dropFieldReq := func(fieldName string) *milvuspb.AlterCollectionSchemaRequest {
|
|
return &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: fieldName},
|
|
},
|
|
},
|
|
},
|
|
}
|
|
}
|
|
|
|
// field not found
|
|
resp, err = core.AlterCollectionSchema(ctx, dropFieldReq("nonexistent"))
|
|
require.ErrorIs(t, merr.CheckRPCCall(resp.GetAlterStatus(), err), merr.ErrParameterInvalid)
|
|
|
|
// drop field2 successfully
|
|
resp, err = core.AlterCollectionSchema(ctx, dropFieldReq("field2"))
|
|
require.NoError(t, merr.CheckRPCCall(resp.GetAlterStatus(), err))
|
|
assertFieldNotExists(t, ctx, core, dbName, collectionName, "field2")
|
|
assertFieldExists(t, ctx, core, dbName, collectionName, "field1", 100)
|
|
assertFieldExists(t, ctx, core, dbName, collectionName, "field3", 102)
|
|
assertSchemaVersion(t, ctx, core, dbName, collectionName, 3)
|
|
assertMaxFieldIDProperty(t, ctx, core, dbName, collectionName, 102)
|
|
|
|
// add field4 after drop: fieldID should not reuse 101 (the dropped field2's ID)
|
|
addResp, err = core.AddCollectionField(ctx, &milvuspb.AddCollectionFieldRequest{
|
|
DbName: dbName,
|
|
CollectionName: collectionName,
|
|
Schema: getFieldSchema("field4"),
|
|
})
|
|
require.NoError(t, merr.CheckRPCCall(addResp, err))
|
|
assertFieldExists(t, ctx, core, dbName, collectionName, "field4", 103)
|
|
assertSchemaVersion(t, ctx, core, dbName, collectionName, 4)
|
|
assertMaxFieldIDProperty(t, ctx, core, dbName, collectionName, 103)
|
|
|
|
// drop field3 successfully
|
|
resp, err = core.AlterCollectionSchema(ctx, dropFieldReq("field3"))
|
|
require.NoError(t, merr.CheckRPCCall(resp.GetAlterStatus(), err))
|
|
assertFieldNotExists(t, ctx, core, dbName, collectionName, "field3")
|
|
assertFieldExists(t, ctx, core, dbName, collectionName, "field1", 100)
|
|
assertFieldExists(t, ctx, core, dbName, collectionName, "field4", 103)
|
|
assertSchemaVersion(t, ctx, core, dbName, collectionName, 5)
|
|
assertMaxFieldIDProperty(t, ctx, core, dbName, collectionName, 103)
|
|
}
|
|
|
|
func TestDDLCallbacksAlterCollectionDropFieldWaitsForSchemaDropReady(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)
|
|
addResp, err := core.AddCollectionField(ctx, &milvuspb.AddCollectionFieldRequest{
|
|
DbName: dbName,
|
|
CollectionName: collectionName,
|
|
Schema: getFieldSchema("field2"),
|
|
})
|
|
require.NoError(t, merr.CheckRPCCall(addResp, err))
|
|
assertFieldExists(t, ctx, core, dbName, collectionName, "field2", 101)
|
|
assertSchemaVersion(t, ctx, core, dbName, collectionName, 1)
|
|
|
|
barrierErr := errors.New("proxy version barrier")
|
|
b := mock_balancer.NewMockBalancer(t)
|
|
b.EXPECT().WaitUntilWALbasedDDLReady(mock.Anything).Return(nil).Maybe()
|
|
b.EXPECT().WaitUntilSchemaDropReady(mock.Anything).Return(barrierErr).Once()
|
|
b.EXPECT().Close().Return().Maybe()
|
|
balance.ResetBalancer()
|
|
balance.Register(b)
|
|
|
|
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: "field2"},
|
|
},
|
|
},
|
|
},
|
|
})
|
|
require.Error(t, merr.CheckRPCCall(resp.GetAlterStatus(), err))
|
|
require.Contains(t, resp.GetAlterStatus().GetDetail(), "failed to wait until schema drop ready")
|
|
assertFieldExists(t, ctx, core, dbName, collectionName, "field2", 101)
|
|
assertSchemaVersion(t, ctx, core, dbName, collectionName, 1)
|
|
}
|
|
|
|
func TestDDLCallbacksAlterCollectionSchemaAddSkipsSchemaDropReady(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)
|
|
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"},
|
|
},
|
|
}
|
|
varcharBytes, err := proto.Marshal(varcharFieldSchema)
|
|
require.NoError(t, err)
|
|
addFieldResp, err := core.AddCollectionField(ctx, &milvuspb.AddCollectionFieldRequest{
|
|
DbName: dbName,
|
|
CollectionName: collectionName,
|
|
Schema: varcharBytes,
|
|
})
|
|
require.NoError(t, merr.CheckRPCCall(addFieldResp, err))
|
|
assertSchemaVersion(t, ctx, core, dbName, collectionName, 1)
|
|
|
|
b := mock_balancer.NewMockBalancer(t)
|
|
b.EXPECT().WaitUntilWALbasedDDLReady(mock.Anything).Return(nil).Maybe()
|
|
b.EXPECT().Close().Return().Maybe()
|
|
balance.ResetBalancer()
|
|
balance.Register(b)
|
|
|
|
resp, err := core.AlterCollectionSchema(ctx, buildAlterSchemaReq(dbName, collectionName, "text_input", "sparse_output", "bm25_fn"))
|
|
require.NoError(t, merr.CheckRPCCall(resp.GetAlterStatus(), err))
|
|
assertSchemaVersion(t, ctx, core, dbName, collectionName, 2)
|
|
}
|
|
|
|
func assertFieldNotExists(t *testing.T, ctx context.Context, core *Core, dbName string, collectionName string, fieldName string) {
|
|
coll, err := core.meta.GetCollectionByName(ctx, dbName, collectionName, typeutil.MaxTimestamp, false)
|
|
require.NoError(t, err)
|
|
for _, field := range coll.Fields {
|
|
if field.Name == fieldName {
|
|
require.Fail(t, "field should not exist", "field %s still exists", fieldName)
|
|
}
|
|
}
|
|
}
|
|
|
|
func assertMaxFieldIDProperty(t *testing.T, ctx context.Context, core *Core, dbName string, collectionName string, expectedMaxFieldID int64) {
|
|
coll, err := core.meta.GetCollectionByName(ctx, dbName, collectionName, typeutil.MaxTimestamp, false)
|
|
require.NoError(t, err)
|
|
for _, kv := range coll.Properties {
|
|
if kv.Key == common.MaxFieldIDKey {
|
|
require.Equal(t, expectedMaxFieldID, mustParseInt64(kv.Value))
|
|
return
|
|
}
|
|
}
|
|
require.Fail(t, "max_field_id property not found")
|
|
}
|
|
|
|
func mustParseInt64(s string) int64 {
|
|
v, err := strconv.ParseInt(s, 10, 64)
|
|
if err != nil {
|
|
panic(err)
|
|
}
|
|
return v
|
|
}
|
|
|
|
func TestBuildSchemaForDropFunctionField(t *testing.T) {
|
|
t.Run("function not found", func(t *testing.T) {
|
|
coll := &model.Collection{
|
|
Functions: []*model.Function{
|
|
{Name: "func1", OutputFieldIDs: []int64{103}},
|
|
},
|
|
}
|
|
_, _, _, err := buildSchemaForDropFunctionField(coll, "nonexistent")
|
|
require.Error(t, err)
|
|
require.Contains(t, err.Error(), "function not found")
|
|
})
|
|
|
|
t.Run("dropping the last vector field rejected", func(t *testing.T) {
|
|
coll := &model.Collection{
|
|
Fields: []*model.Field{
|
|
{FieldID: 100, Name: "pk", DataType: schemapb.DataType_Int64, IsPrimaryKey: true},
|
|
{FieldID: 101, Name: "text", DataType: schemapb.DataType_VarChar},
|
|
{FieldID: 102, Name: "only_vec", DataType: schemapb.DataType_FloatVector, IsFunctionOutput: true},
|
|
},
|
|
Functions: []*model.Function{
|
|
{Name: "embed_func", Type: schemapb.FunctionType_TextEmbedding, InputFieldIDs: []int64{101}, OutputFieldIDs: []int64{102}, OutputFieldNames: []string{"only_vec"}},
|
|
},
|
|
SchemaVersion: 1,
|
|
}
|
|
_, _, _, err := buildSchemaForDropFunctionField(coll, "embed_func")
|
|
require.ErrorContains(t, err, "leave no vector field")
|
|
})
|
|
|
|
t.Run("text embedding function can be dropped", func(t *testing.T) {
|
|
coll := &model.Collection{
|
|
Name: "test_coll",
|
|
Fields: []*model.Field{
|
|
{FieldID: 100, Name: "pk"},
|
|
{FieldID: 101, Name: "text"},
|
|
{FieldID: 102, Name: "keep_vec"},
|
|
{FieldID: 103, Name: "dense_vec", IsFunctionOutput: true},
|
|
},
|
|
Functions: []*model.Function{
|
|
{Name: "embed_func", Type: schemapb.FunctionType_TextEmbedding, InputFieldIDs: []int64{101}, OutputFieldIDs: []int64{103}, OutputFieldNames: []string{"dense_vec"}},
|
|
},
|
|
SchemaVersion: 2,
|
|
}
|
|
schema, _, droppedFieldIDs, err := buildSchemaForDropFunctionField(coll, "embed_func")
|
|
require.NoError(t, err)
|
|
require.Equal(t, []int64{103}, droppedFieldIDs)
|
|
require.Equal(t, 0, len(schema.Functions))
|
|
for _, f := range schema.Fields {
|
|
require.NotEqual(t, "dense_vec", f.Name)
|
|
}
|
|
})
|
|
|
|
t.Run("drop function removes function and output fields", func(t *testing.T) {
|
|
coll := &model.Collection{
|
|
Name: "test_coll",
|
|
Fields: []*model.Field{
|
|
{FieldID: 100, Name: "pk"},
|
|
{FieldID: 101, Name: "text"},
|
|
{FieldID: 102, Name: "vec"},
|
|
{FieldID: 103, Name: "embedding_vec"},
|
|
},
|
|
Functions: []*model.Function{
|
|
{
|
|
Name: "embedding_func",
|
|
Type: schemapb.FunctionType_BM25,
|
|
InputFieldIDs: []int64{101},
|
|
InputFieldNames: []string{"text"},
|
|
OutputFieldIDs: []int64{103},
|
|
OutputFieldNames: []string{"embedding_vec"},
|
|
},
|
|
},
|
|
SchemaVersion: 5,
|
|
}
|
|
|
|
schema, properties, droppedFieldIDs, err := buildSchemaForDropFunctionField(coll, "embedding_func")
|
|
require.NoError(t, err)
|
|
require.Equal(t, []int64{103}, droppedFieldIDs)
|
|
|
|
// output field removed
|
|
require.Equal(t, 3, len(schema.Fields))
|
|
for _, f := range schema.Fields {
|
|
require.NotEqual(t, "embedding_vec", f.Name)
|
|
}
|
|
|
|
// function removed
|
|
require.Equal(t, 0, len(schema.Functions))
|
|
|
|
// max_field_id updated
|
|
require.NotNil(t, properties)
|
|
|
|
// schema version incremented
|
|
require.Equal(t, int32(6), schema.Version)
|
|
})
|
|
|
|
t.Run("preserves input fields and other functions", func(t *testing.T) {
|
|
coll := &model.Collection{
|
|
Name: "test_coll",
|
|
Fields: []*model.Field{
|
|
{FieldID: 100, Name: "pk"},
|
|
{FieldID: 101, Name: "text"},
|
|
{FieldID: 102, Name: "sparse_vec"},
|
|
{FieldID: 103, Name: "dense_vec"},
|
|
},
|
|
Functions: []*model.Function{
|
|
{
|
|
Name: "bm25_func",
|
|
Type: schemapb.FunctionType_BM25,
|
|
InputFieldIDs: []int64{101},
|
|
OutputFieldIDs: []int64{102},
|
|
OutputFieldNames: []string{"sparse_vec"},
|
|
},
|
|
{
|
|
Name: "embed_func",
|
|
Type: schemapb.FunctionType_TextEmbedding,
|
|
InputFieldIDs: []int64{101},
|
|
OutputFieldIDs: []int64{103},
|
|
OutputFieldNames: []string{"dense_vec"},
|
|
},
|
|
},
|
|
SchemaVersion: 3,
|
|
}
|
|
|
|
schema, _, droppedFieldIDs, err := buildSchemaForDropFunctionField(coll, "bm25_func")
|
|
require.NoError(t, err)
|
|
require.Equal(t, []int64{102}, droppedFieldIDs)
|
|
|
|
// only sparse_vec removed, text and dense_vec preserved
|
|
fieldNames := make([]string, 0, len(schema.Fields))
|
|
for _, f := range schema.Fields {
|
|
fieldNames = append(fieldNames, f.Name)
|
|
}
|
|
require.Contains(t, fieldNames, "pk")
|
|
require.Contains(t, fieldNames, "text")
|
|
require.Contains(t, fieldNames, "dense_vec")
|
|
require.NotContains(t, fieldNames, "sparse_vec")
|
|
|
|
// only bm25_func removed, embed_func preserved
|
|
require.Equal(t, 1, len(schema.Functions))
|
|
require.Equal(t, "embed_func", schema.Functions[0].Name)
|
|
})
|
|
|
|
t.Run("corrupt persisted output ids rejected (primary key protection)", func(t *testing.T) {
|
|
// Stored with output name "sparse_vec" (id 102) but a stale/injected pk id
|
|
// (100) in OutputFieldIDs; drop must re-resolve from the name and reject the
|
|
// mismatch instead of deleting the primary key past the name-based guard.
|
|
coll := &model.Collection{
|
|
Name: "test_coll",
|
|
Fields: []*model.Field{
|
|
{FieldID: 100, Name: "pk", DataType: schemapb.DataType_Int64, IsPrimaryKey: true},
|
|
{FieldID: 101, Name: "text", DataType: schemapb.DataType_VarChar},
|
|
{FieldID: 102, Name: "sparse_vec", DataType: schemapb.DataType_SparseFloatVector},
|
|
{FieldID: 103, Name: "keep_vec", DataType: schemapb.DataType_FloatVector},
|
|
},
|
|
Functions: []*model.Function{
|
|
{
|
|
Name: "bm25_func", Type: schemapb.FunctionType_BM25,
|
|
InputFieldIDs: []int64{101},
|
|
OutputFieldIDs: []int64{100, 102}, // 100 = pk injected
|
|
OutputFieldNames: []string{"sparse_vec"},
|
|
},
|
|
},
|
|
SchemaVersion: 3,
|
|
}
|
|
_, _, _, err := buildSchemaForDropFunctionField(coll, "bm25_func")
|
|
require.ErrorContains(t, err, "do not align with output field names")
|
|
})
|
|
}
|
|
|
|
func TestBuildSchemaForDropField(t *testing.T) {
|
|
baseColl := func() *model.Collection {
|
|
return &model.Collection{
|
|
Name: "test_coll",
|
|
Fields: []*model.Field{
|
|
{FieldID: 100, Name: "pk"},
|
|
{FieldID: 101, Name: "vec"},
|
|
{FieldID: 102, Name: "extra"},
|
|
},
|
|
SchemaVersion: 3,
|
|
}
|
|
}
|
|
|
|
t.Run("drop by field name", func(t *testing.T) {
|
|
schema, properties, droppedFieldIDs, err := buildSchemaForDropField(baseColl(), "extra", 0)
|
|
require.NoError(t, err)
|
|
require.Equal(t, 2, len(schema.Fields))
|
|
require.NotNil(t, properties)
|
|
require.Equal(t, []int64{102}, droppedFieldIDs)
|
|
require.Equal(t, int32(4), schema.Version)
|
|
})
|
|
|
|
t.Run("field referenced by function rejected", func(t *testing.T) {
|
|
coll := baseColl()
|
|
coll.Functions = []*model.Function{
|
|
{Name: "bm25", InputFieldNames: []string{"extra"}, OutputFieldNames: []string{"vec"}},
|
|
}
|
|
_, _, _, err := buildSchemaForDropField(coll, "extra", 0)
|
|
require.ErrorContains(t, err, "referenced by function bm25 as input, drop function first")
|
|
|
|
_, _, _, err = buildSchemaForDropField(coll, "vec", 0)
|
|
require.ErrorContains(t, err, "referenced by function bm25 as output, drop function first")
|
|
})
|
|
|
|
t.Run("drop by field id", func(t *testing.T) {
|
|
schema, _, droppedFieldIDs, err := buildSchemaForDropField(baseColl(), "", 101)
|
|
require.NoError(t, err)
|
|
require.Equal(t, []int64{101}, droppedFieldIDs)
|
|
require.Equal(t, 2, len(schema.Fields))
|
|
// remaining fields should be pk and extra
|
|
fieldNames := make([]string, 0, len(schema.Fields))
|
|
for _, f := range schema.Fields {
|
|
fieldNames = append(fieldNames, f.Name)
|
|
}
|
|
require.Contains(t, fieldNames, "pk")
|
|
require.Contains(t, fieldNames, "extra")
|
|
})
|
|
|
|
t.Run("field not found by name", func(t *testing.T) {
|
|
_, _, _, err := buildSchemaForDropField(baseColl(), "nonexistent", 0)
|
|
require.Error(t, err)
|
|
require.Contains(t, err.Error(), "field not found: nonexistent")
|
|
})
|
|
|
|
t.Run("field not found by id", func(t *testing.T) {
|
|
_, _, _, err := buildSchemaForDropField(baseColl(), "", 999)
|
|
require.Error(t, err)
|
|
require.Contains(t, err.Error(), "field not found with id: 999")
|
|
})
|
|
|
|
t.Run("max_field_id property updated", func(t *testing.T) {
|
|
coll := baseColl()
|
|
coll.Properties = []*commonpb.KeyValuePair{
|
|
{Key: common.MaxFieldIDKey, Value: "102"},
|
|
}
|
|
_, properties, _, err := buildSchemaForDropField(coll, "extra", 0)
|
|
require.NoError(t, err)
|
|
var found bool
|
|
for _, kv := range properties {
|
|
if kv.Key == common.MaxFieldIDKey {
|
|
require.Equal(t, "102", kv.Value)
|
|
found = true
|
|
}
|
|
}
|
|
require.True(t, found)
|
|
})
|
|
|
|
collWithStruct := func() *model.Collection {
|
|
return &model.Collection{
|
|
Name: "test_coll",
|
|
Fields: []*model.Field{
|
|
{FieldID: 100, Name: "pk"},
|
|
{FieldID: 101, Name: "vec"},
|
|
},
|
|
StructArrayFields: []*model.StructArrayField{
|
|
{
|
|
FieldID: 102, Name: "paragraphs",
|
|
Fields: []*model.Field{
|
|
{FieldID: 103, Name: "para_text"},
|
|
{FieldID: 104, Name: "para_embed"},
|
|
},
|
|
},
|
|
},
|
|
SchemaVersion: 3,
|
|
}
|
|
}
|
|
|
|
t.Run("drop whole struct array field by name", func(t *testing.T) {
|
|
schema, _, droppedFieldIDs, err := buildSchemaForDropField(collWithStruct(), "paragraphs", 0)
|
|
require.NoError(t, err)
|
|
require.Equal(t, 2, len(schema.Fields)) // pk + vec unchanged
|
|
require.Equal(t, 0, len(schema.StructArrayFields)) // struct removed
|
|
require.Equal(t, []int64{102, 103, 104}, droppedFieldIDs)
|
|
require.Equal(t, int32(4), schema.Version)
|
|
})
|
|
|
|
t.Run("drop whole struct array field by id", func(t *testing.T) {
|
|
schema, _, droppedFieldIDs, err := buildSchemaForDropField(collWithStruct(), "", 102)
|
|
require.NoError(t, err)
|
|
require.Equal(t, 0, len(schema.StructArrayFields))
|
|
require.Equal(t, []int64{102, 103, 104}, droppedFieldIDs)
|
|
})
|
|
}
|
|
|
|
func TestCascadeDropFieldIndexesInline(t *testing.T) {
|
|
buildResult := func(droppedFieldIDs []int64) message.BroadcastResultAlterCollectionMessageV2 {
|
|
raw := message.NewAlterCollectionMessageBuilderV2().
|
|
WithHeader(&messagespb.AlterCollectionMessageHeader{
|
|
CollectionId: 1,
|
|
DroppedFieldIds: droppedFieldIDs,
|
|
}).
|
|
WithBody(&messagespb.AlterCollectionMessageBody{
|
|
Updates: &messagespb.AlterCollectionMessageUpdates{},
|
|
}).
|
|
WithBroadcast([]string{funcutil.GetControlChannel("test")}).
|
|
MustBuildBroadcast()
|
|
msg := message.MustAsBroadcastAlterCollectionMessageV2(raw)
|
|
return message.BroadcastResultAlterCollectionMessageV2{
|
|
Message: msg,
|
|
Results: map[string]*message.AppendResult{},
|
|
}
|
|
}
|
|
|
|
t.Run("no dropped fields short circuits", func(t *testing.T) {
|
|
c := newTestCore()
|
|
cb := &DDLCallback{Core: c}
|
|
err := cb.cascadeDropFieldIndexesInline(context.Background(), buildResult(nil))
|
|
require.NoError(t, err)
|
|
})
|
|
|
|
t.Run("DescribeIndex returns ErrIndexNotFound", func(t *testing.T) {
|
|
mixc := imocks.NewMixCoord(t)
|
|
mixc.EXPECT().DescribeIndex(mock.Anything, mock.Anything).Return(
|
|
&indexpb.DescribeIndexResponse{Status: merr.Status(merr.WrapErrIndexNotFound("idx"))}, nil,
|
|
)
|
|
c := newTestCore(withMixCoord(mixc))
|
|
cb := &DDLCallback{Core: c}
|
|
err := cb.cascadeDropFieldIndexesInline(context.Background(), buildResult([]int64{101}))
|
|
require.NoError(t, err)
|
|
})
|
|
|
|
t.Run("DescribeIndex returns other error", func(t *testing.T) {
|
|
mixc := imocks.NewMixCoord(t)
|
|
mixc.EXPECT().DescribeIndex(mock.Anything, mock.Anything).Return(
|
|
nil, errors.New("rpc unavailable"),
|
|
)
|
|
c := newTestCore(withMixCoord(mixc))
|
|
cb := &DDLCallback{Core: c}
|
|
err := cb.cascadeDropFieldIndexesInline(context.Background(), buildResult([]int64{101}))
|
|
require.Error(t, err)
|
|
// Match the stable prefix of the DescribeIndex failure path. The
|
|
// "for cascade drop" suffix is incidental and may be reworded without
|
|
// changing behavior, so we don't assert on it.
|
|
require.Contains(t, err.Error(), "failed to describe indexes")
|
|
require.ErrorContains(t, err, "rpc unavailable")
|
|
})
|
|
|
|
t.Run("no matching indexes for dropped field", func(t *testing.T) {
|
|
mixc := imocks.NewMixCoord(t)
|
|
mixc.EXPECT().DescribeIndex(mock.Anything, mock.Anything).Return(
|
|
&indexpb.DescribeIndexResponse{
|
|
Status: merr.Success(),
|
|
IndexInfos: []*indexpb.IndexInfo{
|
|
{FieldID: 200, IndexID: 1, IndexName: "idx_other"},
|
|
},
|
|
}, nil,
|
|
)
|
|
c := newTestCore(withMixCoord(mixc))
|
|
cb := &DDLCallback{Core: c}
|
|
err := cb.cascadeDropFieldIndexesInline(context.Background(), buildResult([]int64{101}))
|
|
require.NoError(t, err)
|
|
})
|
|
}
|
|
|
|
func TestAlterCollectionV2AckCallbackUsesHeaderDroppedFieldIDs(t *testing.T) {
|
|
raw := message.NewAlterCollectionMessageBuilderV2().
|
|
WithHeader(&messagespb.AlterCollectionMessageHeader{
|
|
CollectionId: 1,
|
|
UpdateMask: &fieldmaskpb.FieldMask{
|
|
Paths: []string{message.FieldMaskCollectionSchema},
|
|
},
|
|
CacheExpirations: &messagespb.CacheExpirations{},
|
|
DroppedFieldIds: []int64{101},
|
|
}).
|
|
WithBody(&messagespb.AlterCollectionMessageBody{
|
|
Updates: &messagespb.AlterCollectionMessageUpdates{
|
|
Schema: &schemapb.CollectionSchema{
|
|
Name: "test",
|
|
Version: 2,
|
|
},
|
|
},
|
|
}).
|
|
WithBroadcast([]string{funcutil.GetControlChannel("test")}).
|
|
MustBuildBroadcast()
|
|
|
|
meta := &mockMetaTable{}
|
|
meta.AlterCollectionFunc = func(ctx context.Context, result message.BroadcastResultAlterCollectionMessageV2) error {
|
|
return nil
|
|
}
|
|
|
|
mixc := imocks.NewMixCoord(t)
|
|
mixc.EXPECT().DescribeIndex(mock.Anything, mock.Anything).Return(
|
|
&indexpb.DescribeIndexResponse{
|
|
Status: merr.Success(),
|
|
IndexInfos: []*indexpb.IndexInfo{
|
|
{FieldID: 200, IndexID: 1, IndexName: "idx_other"},
|
|
},
|
|
}, nil,
|
|
)
|
|
|
|
broker := newValidMockBroker()
|
|
c := newTestCore(withMeta(meta), withMixCoord(mixc), withBroker(broker))
|
|
cb := &DDLCallback{Core: c}
|
|
err := cb.alterCollectionV2AckCallback(context.Background(), message.BroadcastResultAlterCollectionMessageV2{
|
|
Message: message.MustAsBroadcastAlterCollectionMessageV2(raw),
|
|
Results: map[string]*message.AppendResult{
|
|
funcutil.GetControlChannel("test"): {TimeTick: 100},
|
|
},
|
|
})
|
|
require.NoError(t, err)
|
|
}
|
|
|
|
func newBoundIndexTestPlan() (*model.Collection, *schemautil.AlterSchemaAddPlan) {
|
|
coll := &model.Collection{CollectionID: 1, DBID: 1, Name: "coll"}
|
|
plan := &schemautil.AlterSchemaAddPlan{
|
|
Kind: schemautil.AlterSchemaAddFunctionField,
|
|
Field: &schemapb.FieldSchema{
|
|
FieldID: 101,
|
|
Name: "sparse",
|
|
DataType: schemapb.DataType_SparseFloatVector,
|
|
},
|
|
Function: &schemapb.FunctionSchema{Name: "bm25_fn", Type: schemapb.FunctionType_BM25},
|
|
IndexExtraParams: []*commonpb.KeyValuePair{
|
|
{Key: common.IndexTypeKey, Value: "SPARSE_INVERTED_INDEX"},
|
|
{Key: common.MetricTypeKey, Value: "BM25"},
|
|
},
|
|
}
|
|
return coll, plan
|
|
}
|
|
|
|
func withNoIndexMixCoord(t *testing.T) *imocks.MixCoord {
|
|
mixc := imocks.NewMixCoord(t)
|
|
mixc.EXPECT().DescribeIndex(mock.Anything, mock.Anything).Return(
|
|
&indexpb.DescribeIndexResponse{Status: merr.Status(merr.WrapErrIndexNotFound("sparse"))}, nil,
|
|
).Maybe()
|
|
return mixc
|
|
}
|
|
|
|
func withOkTso(t *testing.T) tso.Allocator {
|
|
alloc := mocktso.NewAllocator(t)
|
|
alloc.EXPECT().GenerateTSO(mock.Anything).Return(uint64(100), nil).Maybe()
|
|
return alloc
|
|
}
|
|
|
|
func TestPrepareBoundFieldIndex(t *testing.T) {
|
|
t.Run("happy path materializes replay-deterministic index", func(t *testing.T) {
|
|
coll, plan := newBoundIndexTestPlan()
|
|
c := newTestCore(withMixCoord(withNoIndexMixCoord(t)), withValidIDAllocator(), withTsoAllocator(withOkTso(t)))
|
|
fieldIndex, err := c.prepareBoundFieldIndex(context.Background(), coll, plan)
|
|
require.NoError(t, err)
|
|
info := fieldIndex.GetIndexInfo()
|
|
require.Equal(t, int64(101), info.GetFieldID())
|
|
require.Equal(t, "sparse", info.GetIndexName())
|
|
require.Positive(t, info.GetIndexID())
|
|
params := funcutil.KeyValuePair2Map(info.GetIndexParams())
|
|
require.Equal(t, "SPARSE_INVERTED_INDEX", params[common.IndexTypeKey])
|
|
require.Equal(t, "1.2", params["bm25_k1"])
|
|
// UserIndexParams keep the user's ORIGINAL params (create_index
|
|
// convention), so a later create_index with identical params is
|
|
// treated as idempotent instead of a distinct index.
|
|
require.Equal(t, plan.IndexExtraParams, info.GetUserIndexParams())
|
|
})
|
|
|
|
t.Run("index name conflict rejected", func(t *testing.T) {
|
|
coll, plan := newBoundIndexTestPlan()
|
|
mixc := imocks.NewMixCoord(t)
|
|
mixc.EXPECT().DescribeIndex(mock.Anything, mock.Anything).Return(
|
|
&indexpb.DescribeIndexResponse{
|
|
Status: merr.Success(),
|
|
IndexInfos: []*indexpb.IndexInfo{{FieldID: 100, IndexID: 1, IndexName: "sparse"}},
|
|
}, nil,
|
|
)
|
|
c := newTestCore(withMixCoord(mixc), withValidIDAllocator(), withTsoAllocator(withOkTso(t)))
|
|
_, err := c.prepareBoundFieldIndex(context.Background(), coll, plan)
|
|
require.ErrorIs(t, err, merr.ErrParameterInvalid)
|
|
require.ErrorContains(t, err, "already exists")
|
|
})
|
|
|
|
t.Run("describe index transient error rejects the DDL", func(t *testing.T) {
|
|
coll, plan := newBoundIndexTestPlan()
|
|
mixc := imocks.NewMixCoord(t)
|
|
mixc.EXPECT().DescribeIndex(mock.Anything, mock.Anything).Return(nil, errors.New("rpc unavailable"))
|
|
c := newTestCore(withMixCoord(mixc), withValidIDAllocator(), withTsoAllocator(withOkTso(t)))
|
|
_, err := c.prepareBoundFieldIndex(context.Background(), coll, plan)
|
|
require.Error(t, err)
|
|
require.ErrorContains(t, err, "failed to list existing indexes")
|
|
})
|
|
|
|
t.Run("index id allocation failure", func(t *testing.T) {
|
|
coll, plan := newBoundIndexTestPlan()
|
|
c := newTestCore(withMixCoord(withNoIndexMixCoord(t)), withInvalidIDAllocator(), withTsoAllocator(withOkTso(t)))
|
|
_, err := c.prepareBoundFieldIndex(context.Background(), coll, plan)
|
|
require.Error(t, err)
|
|
require.ErrorContains(t, err, "failed to allocate index id")
|
|
})
|
|
|
|
t.Run("incompatible index type for field data type rejected", func(t *testing.T) {
|
|
coll, plan := newBoundIndexTestPlan()
|
|
plan.Field = &schemapb.FieldSchema{
|
|
FieldID: 102,
|
|
Name: "binary_mh",
|
|
DataType: schemapb.DataType_BinaryVector,
|
|
TypeParams: []*commonpb.KeyValuePair{
|
|
{Key: common.DimKey, Value: "512"},
|
|
},
|
|
}
|
|
plan.Function = &schemapb.FunctionSchema{Name: "mh_fn", Type: schemapb.FunctionType_MinHash}
|
|
// SPARSE index on a binary vector field: must be rejected pre-broadcast.
|
|
c := newTestCore(withMixCoord(withNoIndexMixCoord(t)), withValidIDAllocator(), withTsoAllocator(withOkTso(t)))
|
|
_, err := c.prepareBoundFieldIndex(context.Background(), coll, plan)
|
|
require.Error(t, err)
|
|
})
|
|
|
|
t.Run("dimension mismatch rejected", func(t *testing.T) {
|
|
coll, plan := newBoundIndexTestPlan()
|
|
plan.Field = &schemapb.FieldSchema{
|
|
FieldID: 102,
|
|
Name: "binary_mh",
|
|
DataType: schemapb.DataType_BinaryVector,
|
|
TypeParams: []*commonpb.KeyValuePair{
|
|
{Key: common.DimKey, Value: "512"},
|
|
},
|
|
}
|
|
plan.Function = &schemapb.FunctionSchema{Name: "mh_fn", Type: schemapb.FunctionType_MinHash}
|
|
plan.IndexExtraParams = []*commonpb.KeyValuePair{
|
|
{Key: common.IndexTypeKey, Value: "MINHASH_LSH"},
|
|
{Key: common.MetricTypeKey, Value: "MHJACCARD"},
|
|
{Key: common.DimKey, Value: "1024"},
|
|
}
|
|
c := newTestCore(withMixCoord(withNoIndexMixCoord(t)), withValidIDAllocator(), withTsoAllocator(withOkTso(t)))
|
|
_, err := c.prepareBoundFieldIndex(context.Background(), coll, plan)
|
|
require.Error(t, err)
|
|
require.ErrorContains(t, err, "dimension mismatch")
|
|
})
|
|
|
|
t.Run("bogus warmup policy rejected", func(t *testing.T) {
|
|
coll, plan := newBoundIndexTestPlan()
|
|
plan.IndexExtraParams = append(plan.IndexExtraParams, &commonpb.KeyValuePair{
|
|
Key: common.WarmupKey, Value: "bogus",
|
|
})
|
|
c := newTestCore(withMixCoord(withNoIndexMixCoord(t)), withValidIDAllocator(), withTsoAllocator(withOkTso(t)))
|
|
_, err := c.prepareBoundFieldIndex(context.Background(), coll, plan)
|
|
require.Error(t, err)
|
|
require.ErrorContains(t, err, "warmup")
|
|
})
|
|
|
|
t.Run("malformed index name rejected", func(t *testing.T) {
|
|
coll, plan := newBoundIndexTestPlan()
|
|
plan.IndexName = "1bad-name"
|
|
c := newTestCore(withMixCoord(withNoIndexMixCoord(t)), withValidIDAllocator(), withTsoAllocator(withOkTso(t)))
|
|
_, err := c.prepareBoundFieldIndex(context.Background(), coll, plan)
|
|
require.Error(t, err)
|
|
require.ErrorContains(t, err, "Invalid index name")
|
|
})
|
|
|
|
t.Run("oversized index params rejected", func(t *testing.T) {
|
|
coll, plan := newBoundIndexTestPlan()
|
|
plan.IndexExtraParams = append(plan.IndexExtraParams, &commonpb.KeyValuePair{
|
|
Key: "huge",
|
|
Value: strings.Repeat("x", paramtable.Get().ProxyCfg.MaxIndexParamsSize.GetAsInt()+1),
|
|
})
|
|
c := newTestCore(withMixCoord(withNoIndexMixCoord(t)), withValidIDAllocator(), withTsoAllocator(withOkTso(t)))
|
|
_, err := c.prepareBoundFieldIndex(context.Background(), coll, plan)
|
|
require.Error(t, err)
|
|
require.ErrorContains(t, err, "exceeds limit")
|
|
})
|
|
|
|
t.Run("tso allocation failure", func(t *testing.T) {
|
|
coll, plan := newBoundIndexTestPlan()
|
|
alloc := mocktso.NewAllocator(t)
|
|
alloc.EXPECT().GenerateTSO(mock.Anything).Return(uint64(0), errors.New("tso unavailable"))
|
|
c := newTestCore(withMixCoord(withNoIndexMixCoord(t)), withValidIDAllocator(), withTsoAllocator(alloc))
|
|
_, err := c.prepareBoundFieldIndex(context.Background(), coll, plan)
|
|
require.Error(t, err)
|
|
require.ErrorContains(t, err, "failed to allocate timestamp")
|
|
})
|
|
}
|
|
|
|
func TestApplyBoundFieldIndexesInline(t *testing.T) {
|
|
wal := mock_streaming.NewMockWALAccesser(t)
|
|
wal.EXPECT().ControlChannel().Return(funcutil.GetControlChannel("by-dev-rootcoord-dml_0")).Maybe()
|
|
streaming.SetWALForTest(wal)
|
|
|
|
boundFieldIndex := &indexpb.FieldIndex{
|
|
IndexInfo: &indexpb.IndexInfo{
|
|
CollectionID: 1,
|
|
FieldID: 101,
|
|
IndexID: 201,
|
|
IndexName: "sparse",
|
|
IndexParams: []*commonpb.KeyValuePair{
|
|
{Key: common.IndexTypeKey, Value: "SPARSE_INVERTED_INDEX"},
|
|
},
|
|
},
|
|
}
|
|
|
|
buildResult := func(bound []*indexpb.FieldIndex) message.BroadcastResultAlterCollectionMessageV2 {
|
|
raw := message.NewAlterCollectionMessageBuilderV2().
|
|
WithHeader(&messagespb.AlterCollectionMessageHeader{
|
|
DbId: 1,
|
|
CollectionId: 1,
|
|
}).
|
|
WithBody(&messagespb.AlterCollectionMessageBody{
|
|
Updates: &messagespb.AlterCollectionMessageUpdates{
|
|
BoundFieldIndexes: bound,
|
|
},
|
|
}).
|
|
WithBroadcast([]string{funcutil.GetControlChannel("by-dev-rootcoord-dml_0")}).
|
|
MustBuildBroadcast()
|
|
msg := message.MustAsBroadcastAlterCollectionMessageV2(raw.WithBroadcastID(1))
|
|
return message.BroadcastResultAlterCollectionMessageV2{
|
|
Message: msg,
|
|
Results: map[string]*message.AppendResult{},
|
|
}
|
|
}
|
|
|
|
t.Run("no bound indexes short circuits", func(t *testing.T) {
|
|
registry.ResetRegistration()
|
|
cb := &DDLCallback{Core: newTestCore()}
|
|
require.NoError(t, cb.applyBoundFieldIndexesInline(context.Background(), buildResult(nil)))
|
|
})
|
|
|
|
t.Run("dispatches synthetic create index message", func(t *testing.T) {
|
|
registry.ResetRegistration()
|
|
var applied []*indexpb.FieldIndex
|
|
registry.RegisterCreateIndexV2AckCallback(func(ctx context.Context, result message.BroadcastResultCreateIndexMessageV2) error {
|
|
applied = append(applied, result.Message.MustBody().GetFieldIndex())
|
|
return nil
|
|
})
|
|
cb := &DDLCallback{Core: newTestCore()}
|
|
require.NoError(t, cb.applyBoundFieldIndexesInline(context.Background(), buildResult([]*indexpb.FieldIndex{boundFieldIndex})))
|
|
require.Len(t, applied, 1)
|
|
require.Equal(t, int64(201), applied[0].GetIndexInfo().GetIndexID())
|
|
require.Equal(t, "sparse", applied[0].GetIndexInfo().GetIndexName())
|
|
})
|
|
|
|
t.Run("dispatch failure propagates so the ack callback retries", func(t *testing.T) {
|
|
registry.ResetRegistration()
|
|
registry.RegisterCreateIndexV2AckCallback(func(ctx context.Context, result message.BroadcastResultCreateIndexMessageV2) error {
|
|
return errors.New("catalog write failed")
|
|
})
|
|
cb := &DDLCallback{Core: newTestCore()}
|
|
err := cb.applyBoundFieldIndexesInline(context.Background(), buildResult([]*indexpb.FieldIndex{boundFieldIndex}))
|
|
require.Error(t, err)
|
|
require.ErrorContains(t, err, "failed to apply bound field index")
|
|
})
|
|
}
|