milvus/internal/rootcoord/ddl_callbacks_collection_test.go
Zhen Ye 02e2170601
enhance: cherry pick patch of new DDL framework and CDC 2 (#45241)
issue: #43897, #44123
pr: #45224
also pick pr: #45216,#45154,#45033,#45145,#45092,#45058,#45029

enhance: Close channel replicator more gracefully (#45029)

issue: https://github.com/milvus-io/milvus/issues/44123

enhance: Show create time for import job (#45058)

issue: https://github.com/milvus-io/milvus/issues/45056

fix: wal state may be unconsistent after recovering from crash (#45092)

issue: #45088, #45086

- Message on control channel should trigger the checkpoint update.
- LastConfrimedMessageID should be recovered from the minimum of
checkpoint or the LastConfirmedMessageID of uncommitted txn.
- Add more log info for wal debugging.

fix: make ack of broadcaster cannot canceled by client (#45145)

issue: #45141

- make ack of broadcaster cannot canceled by rpc.
- make clone for assignment snapshot of wal balancer.
- add server id for GetReplicateCheckpoint to avoid failure.

enhance: support collection and index with WAL-based DDL framework
(#45033)

issue: #43897

- Part of collection/index related DDL is implemented by WAL-based DDL
framework now.
- Support following message type in wal, CreateCollection,
DropCollection, CreatePartition, DropPartition, CreateIndex, AlterIndex,
DropIndex.
- Part of collection/index related DDL can be synced by new CDC now.
- Refactor some UT for collection/index DDL.
- Add Tombstone scheduler to manage the tombstone GC for collection or
partition meta.
- Move the vchannel allocation into streaming pchannel manager.

enhance: support load/release collection/partition with WAL-based DDL
framework (#45154)

issue: #43897

- Load/Release collection/partition is implemented by WAL-based DDL
framework now.
- Support AlterLoadConfig/DropLoadConfig in wal now.
- Load/Release operation can be synced by new CDC now.
- Refactor some UT for load/release DDL.

enhance: Don't start cdc by default (#45216)

issue: https://github.com/milvus-io/milvus/issues/44123


fix: unrecoverable when replicate from old (#45224)

issue: #44962

---------

Signed-off-by: bigsheeper <yihao.dai@zilliz.com>
Signed-off-by: chyezh <chyezh@outlook.com>
Co-authored-by: yihao.dai <yihao.dai@zilliz.com>
2025-11-04 01:35:33 +08:00

153 lines
5.5 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"
"testing"
"github.com/samber/lo"
"github.com/stretchr/testify/require"
"google.golang.org/protobuf/proto"
"github.com/milvus-io/milvus-proto/go-api/v2/milvuspb"
"github.com/milvus-io/milvus-proto/go-api/v2/schemapb"
"github.com/milvus-io/milvus/internal/metastore/model"
"github.com/milvus-io/milvus/pkg/v2/util/funcutil"
"github.com/milvus-io/milvus/pkg/v2/util/merr"
"github.com/milvus-io/milvus/pkg/v2/util/typeutil"
)
func TestDDLCallbacksCollectionDDL(t *testing.T) {
core := initStreamingSystemAndCore(t)
ctx := context.Background()
dbName := "testDB" + funcutil.RandomString(10)
collectionName := "testCollection" + funcutil.RandomString(10)
partitionName := "testPartition" + funcutil.RandomString(10)
testSchema := &schemapb.CollectionSchema{
Name: collectionName,
Description: "",
AutoID: false,
Fields: []*schemapb.FieldSchema{
{
Name: "field1",
DataType: schemapb.DataType_Int64,
},
},
}
schemaBytes, err := proto.Marshal(testSchema)
require.NoError(t, err)
// drop a collection that db not exist should be ignored.
status, err := core.DropCollection(ctx, &milvuspb.DropCollectionRequest{
DbName: "notExistDB",
CollectionName: collectionName,
})
require.NoError(t, merr.CheckRPCCall(status, err))
// drop a collection that collection not exist should be ignored.
status, err = core.DropCollection(ctx, &milvuspb.DropCollectionRequest{
DbName: dbName,
CollectionName: "notExistCollection",
})
require.NoError(t, merr.CheckRPCCall(status, err))
// create a collection that database not exist should return error.
status, err = core.CreateCollection(ctx, &milvuspb.CreateCollectionRequest{
DbName: "notExistDB",
CollectionName: collectionName,
Schema: schemaBytes,
})
require.Error(t, merr.CheckRPCCall(status, err))
// Test CreateCollection
// create a database and a collection.
status, err = core.CreateDatabase(ctx, &milvuspb.CreateDatabaseRequest{
DbName: dbName,
})
require.NoError(t, merr.CheckRPCCall(status, err))
status, err = core.CreateCollection(ctx, &milvuspb.CreateCollectionRequest{
DbName: dbName,
CollectionName: collectionName,
Schema: schemaBytes,
})
require.NoError(t, merr.CheckRPCCall(status, err))
coll, err := core.meta.GetCollectionByName(ctx, dbName, collectionName, typeutil.MaxTimestamp)
require.NoError(t, err)
require.Equal(t, coll.Name, collectionName)
// create a collection with same schema should be idempotent.
status, err = core.CreateCollection(ctx, &milvuspb.CreateCollectionRequest{
DbName: dbName,
CollectionName: collectionName,
Schema: schemaBytes,
})
require.NoError(t, merr.CheckRPCCall(status, err))
// Test CreatePartition
status, err = core.CreatePartition(ctx, &milvuspb.CreatePartitionRequest{
DbName: dbName,
CollectionName: collectionName,
PartitionName: partitionName,
})
require.NoError(t, merr.CheckRPCCall(status, err))
coll, err = core.meta.GetCollectionByName(ctx, dbName, collectionName, typeutil.MaxTimestamp)
require.NoError(t, err)
require.Len(t, coll.Partitions, 2)
require.Contains(t, lo.Map(coll.Partitions, func(p *model.Partition, _ int) string { return p.PartitionName }), partitionName)
// create a partition with same name should be idempotent.
status, err = core.CreatePartition(ctx, &milvuspb.CreatePartitionRequest{
DbName: dbName,
CollectionName: collectionName,
PartitionName: partitionName,
})
require.NoError(t, merr.CheckRPCCall(status, err))
coll, err = core.meta.GetCollectionByName(ctx, dbName, collectionName, typeutil.MaxTimestamp)
require.NoError(t, err)
require.Len(t, coll.Partitions, 2)
status, err = core.DropPartition(ctx, &milvuspb.DropPartitionRequest{
DbName: dbName,
CollectionName: collectionName,
PartitionName: partitionName,
})
require.NoError(t, merr.CheckRPCCall(status, err))
// drop a partition that partition not exist should be idempotent.
status, err = core.DropPartition(ctx, &milvuspb.DropPartitionRequest{
DbName: dbName,
CollectionName: collectionName,
PartitionName: partitionName,
})
require.NoError(t, merr.CheckRPCCall(status, err))
// Test DropCollection
// drop the collection should be ok.
status, err = core.DropCollection(ctx, &milvuspb.DropCollectionRequest{
DbName: dbName,
CollectionName: collectionName,
})
require.NoError(t, merr.CheckRPCCall(status, err))
_, err = core.meta.GetCollectionByName(ctx, dbName, collectionName, typeutil.MaxTimestamp)
require.Error(t, err)
// drop a dropped collection should be idempotent.
status, err = core.DropCollection(ctx, &milvuspb.DropCollectionRequest{
DbName: dbName,
CollectionName: collectionName,
})
require.NoError(t, merr.CheckRPCCall(status, err))
}