Skip to content

M
milvus

milvus / milvus
10 个月 前同步成功

通知 260
Star 22476
Fork 2472
M
milvus

前往新版Gitcode，体验更适合开发者的 AI 搜索 >>

提交 d1f31c7b 编写于 作者: Q quicksilver 提交者： yefu.chen
浏览文件
操作

Cache Docker volumes in code-checker.yaml 

Signed-off-by: Nquicksilver <zhifeng.zhang@zilliz.com>
上级 41fb2d29
隐藏空白更改
内联 并排
Showing with 25 addition and 688 deletion
.github/workflows/code-checker.yaml
浏览文件 @ d1f31c7b
......@@ -42,6 +42,12 @@ jobs:
steps:
- name: Checkout
uses: actions/checkout@v2
- name: Cache Docker Volumes
uses: actions/cache@v1
with:
path: .docker
key: ubuntu${{ matrix.ubuntu }}-${{ hashFiles('internal/core/**') }}
restore-keys: ubuntu${{ matrix.ubuntu }}-
- name: Dockerfile Lint
uses: reviewdog/action-hadolint@v1
with:
......
internal/reader/collection_replica.go
浏览文件 @ d1f31c7b
......@@ -13,7 +13,6 @@ package reader
import "C"
import (
"github.com/zilliztech/milvus-distributed/internal/proto/internalpb"
"log"
"strconv"
"sync"
......@@ -38,18 +37,13 @@ type collectionReplica interface {
removeCollection(collectionID UniqueID) error
getCollectionByID(collectionID UniqueID) (*Collection, error)
getCollectionByName(collectionName string) (*Collection, error)
hasCollection(collectionID UniqueID) bool
// partition
// Partition tags in different collections are not unique,
// so partition api should specify the target collection.
getPartitionNum(collectionID UniqueID) (int, error)
addPartition(collectionID UniqueID, partitionTag string) error
removePartition(collectionID UniqueID, partitionTag string) error
addPartitionsByCollectionMeta(colMeta *etcdpb.CollectionMeta) error
removePartitionsByCollectionMeta(colMeta *etcdpb.CollectionMeta) error
getPartitionByTag(collectionID UniqueID, partitionTag string) (*Partition, error)
hasPartition(collectionID UniqueID, partitionTag string) bool
// segment
getSegmentNum() int
......@@ -148,31 +142,7 @@ func (colReplica *collectionReplicaImpl) getCollectionByName(collectionName stri
return nil, errors.New("Cannot found collection: " + collectionName)
}
func (colReplica *collectionReplicaImpl) hasCollection(collectionID UniqueID) bool {
colReplica.mu.RLock()
defer colReplica.mu.RUnlock()
for _, col := range colReplica.collections {
if col.ID() == collectionID {
return true
}
}
return false
}
//----------------------------------------------------------------------------------------------------- partition
func (colReplica *collectionReplicaImpl) getPartitionNum(collectionID UniqueID) (int, error) {
collection, err := colReplica.getCollectionByID(collectionID)
if err != nil {
return -1, err
}
colReplica.mu.RLock()
defer colReplica.mu.RUnlock()
return len(collection.partitions), nil
}
func (colReplica *collectionReplicaImpl) addPartition(collectionID UniqueID, partitionTag string) error {
collection, err := colReplica.getCollectionByID(collectionID)
if err != nil {
......@@ -212,61 +182,6 @@ func (colReplica *collectionReplicaImpl) removePartition(collectionID UniqueID,
return nil
}
func (colReplica *collectionReplicaImpl) addPartitionsByCollectionMeta(colMeta *etcdpb.CollectionMeta) error {
if !colReplica.hasCollection(colMeta.ID) {
err := errors.New("Cannot find collection, id = " + strconv.FormatInt(colMeta.ID, 10))
return err
}
pToAdd := make([]string, 0)
for _, partitionTag := range colMeta.PartitionTags {
if !colReplica.hasPartition(colMeta.ID, partitionTag) {
pToAdd = append(pToAdd, partitionTag)
}
}
for _, tag := range pToAdd {
err := colReplica.addPartition(colMeta.ID, tag)
if err != nil {
log.Println(err)
}
}
return nil
}
func (colReplica *collectionReplicaImpl) removePartitionsByCollectionMeta(colMeta *etcdpb.CollectionMeta) error {
col, err := colReplica.getCollectionByID(colMeta.ID)
if err != nil {
return err
}
colReplica.mu.Lock()
pToDel := make([]string, 0)
for _, partition := range col.partitions {
hasPartition := false
for _, tag := range colMeta.PartitionTags {
if partition.partitionTag == tag {
hasPartition = true
}
}
if !hasPartition {
pToDel = append(pToDel, partition.partitionTag)
}
}
colReplica.mu.Unlock()
for _, tag := range pToDel {
err := colReplica.removePartition(col.ID(), tag)
if err != nil {
log.Println(err)
}
}
return nil
}
func (colReplica *collectionReplicaImpl) getPartitionByTag(collectionID UniqueID, partitionTag string) (*Partition, error) {
collection, err := colReplica.getCollectionByID(collectionID)
if err != nil {
......@@ -285,25 +200,6 @@ func (colReplica *collectionReplicaImpl) getPartitionByTag(collectionID UniqueID
return nil, errors.New("cannot find partition, tag = " + partitionTag)
}
func (colReplica *collectionReplicaImpl) hasPartition(collectionID UniqueID, partitionTag string) bool {
collection, err := colReplica.getCollectionByID(collectionID)
if err != nil {
log.Println(err)
return false
}
colReplica.mu.RLock()
defer colReplica.mu.RUnlock()
for _, p := range *collection.Partitions() {
if p.Tag() == partitionTag {
return true
}
}
return false
}
//----------------------------------------------------------------------------------------------------- segment
func (colReplica *collectionReplicaImpl) getSegmentNum() int {
colReplica.mu.RLock()
......@@ -313,9 +209,6 @@ func (colReplica *collectionReplicaImpl) getSegmentNum() int {
}
func (colReplica *collectionReplicaImpl) getSegmentStatistics() *internalpb.QueryNodeSegStats {
colReplica.mu.RLock()
defer colReplica.mu.RUnlock()
var statisticData = make([]*internalpb.SegmentStats, 0)
for segmentID, segment := range colReplica.segments {
......@@ -413,9 +306,6 @@ func (colReplica *collectionReplicaImpl) hasSegment(segmentID UniqueID) bool {
//-----------------------------------------------------------------------------------------------------
func (colReplica *collectionReplicaImpl) freeAll() {
colReplica.mu.Lock()
defer colReplica.mu.Unlock()
for _, seg := range colReplica.segments {
deleteSegment(seg)
}
......
internal/reader/collection_replica_test.go
浏览文件 @ d1f31c7b
......@@ -13,60 +13,7 @@ import (
)
//----------------------------------------------------------------------------------------------------- collection
func TestCollectionReplica_getCollectionNum(t *testing.T) {
ctx := context.Background()
node := NewQueryNode(ctx, 0)
collectionName := "collection0"
fieldVec := schemapb.FieldSchema{
Name: "vec",
DataType: schemapb.DataType_VECTOR_FLOAT,
TypeParams: []*commonpb.KeyValuePair{
{
Key: "dim",
Value: "16",
},
},
}
fieldInt := schemapb.FieldSchema{
Name: "age",
DataType: schemapb.DataType_INT32,
TypeParams: []*commonpb.KeyValuePair{
{
Key: "dim",
Value: "1",
},
},
}
schema := schemapb.CollectionSchema{
Name: collectionName,
Fields: []*schemapb.FieldSchema{
&fieldVec, &fieldInt,
},
}
collectionMeta := etcdpb.CollectionMeta{
ID: UniqueID(0),
Schema: &schema,
CreateTime: Timestamp(0),
SegmentIDs: []UniqueID{0},
PartitionTags: []string{"default"},
}
collectionMetaBlob := proto.MarshalTextString(&collectionMeta)
assert.NotEqual(t, "", collectionMetaBlob)
var err = (*node.replica).addCollection(&collectionMeta, collectionMetaBlob)
assert.NoError(t, err)
assert.Equal(t, (*node.replica).getCollectionNum(), 1)
(*node.replica).freeAll()
}
func TestCollectionReplica_addCollection(t *testing.T) {
func TestColSegContainer_addCollection(t *testing.T) {
ctx := context.Background()
node := NewQueryNode(ctx, 0)
......@@ -119,11 +66,9 @@ func TestCollectionReplica_addCollection(t *testing.T) {
assert.Equal(t, collection.meta.Schema.Name, collectionName)
assert.Equal(t, collection.meta.ID, UniqueID(0))
assert.Equal(t, (*node.replica).getCollectionNum(), 1)
(*node.replica).freeAll()
}
func TestCollectionReplica_removeCollection(t *testing.T) {
func TestColSegContainer_removeCollection(t *testing.T) {
ctx := context.Background()
node := NewQueryNode(ctx, 0)
......@@ -182,11 +127,9 @@ func TestCollectionReplica_removeCollection(t *testing.T) {
err = (*node.replica).removeCollection(collectionID)
assert.NoError(t, err)
assert.Equal(t, (*node.replica).getCollectionNum(), 0)
(*node.replica).freeAll()
}
func TestCollectionReplica_getCollectionByID(t *testing.T) {
func TestColSegContainer_getCollectionByID(t *testing.T) {
ctx := context.Background()
node := NewQueryNode(ctx, 0)
......@@ -246,11 +189,9 @@ func TestCollectionReplica_getCollectionByID(t *testing.T) {
assert.NotNil(t, targetCollection)
assert.Equal(t, targetCollection.meta.Schema.Name, "collection0")
assert.Equal(t, targetCollection.meta.ID, UniqueID(0))
(*node.replica).freeAll()
}
func TestCollectionReplica_getCollectionByName(t *testing.T) {
func TestColSegContainer_getCollectionByName(t *testing.T) {
ctx := context.Background()
node := NewQueryNode(ctx, 0)
......@@ -310,68 +251,10 @@ func TestCollectionReplica_getCollectionByName(t *testing.T) {
assert.NotNil(t, targetCollection)
assert.Equal(t, targetCollection.meta.Schema.Name, "collection0")
assert.Equal(t, targetCollection.meta.ID, UniqueID(0))
(*node.replica).freeAll()
}
func TestCollectionReplica_hasCollection(t *testing.T) {
ctx := context.Background()
node := NewQueryNode(ctx, 0)
collectionName := "collection0"
fieldVec := schemapb.FieldSchema{
Name: "vec",
DataType: schemapb.DataType_VECTOR_FLOAT,
TypeParams: []*commonpb.KeyValuePair{
{
Key: "dim",
Value: "16",
},
},
}
fieldInt := schemapb.FieldSchema{
Name: "age",
DataType: schemapb.DataType_INT32,
TypeParams: []*commonpb.KeyValuePair{
{
Key: "dim",
Value: "1",
},
},
}
schema := schemapb.CollectionSchema{
Name: collectionName,
Fields: []*schemapb.FieldSchema{
&fieldVec, &fieldInt,
},
}
collectionMeta := etcdpb.CollectionMeta{
ID: UniqueID(0),
Schema: &schema,
CreateTime: Timestamp(0),
SegmentIDs: []UniqueID{0},
PartitionTags: []string{"default"},
}
collectionMetaBlob := proto.MarshalTextString(&collectionMeta)
assert.NotEqual(t, "", collectionMetaBlob)
var err = (*node.replica).addCollection(&collectionMeta, collectionMetaBlob)
assert.NoError(t, err)
hasCollection := (*node.replica).hasCollection(UniqueID(0))
assert.Equal(t, hasCollection, true)
hasCollection = (*node.replica).hasCollection(UniqueID(1))
assert.Equal(t, hasCollection, false)
(*node.replica).freeAll()
}
//----------------------------------------------------------------------------------------------------- partition
func TestCollectionReplica_getPartitionNum(t *testing.T) {
func TestColSegContainer_addPartition(t *testing.T) {
ctx := context.Background()
node := NewQueryNode(ctx, 0)
......@@ -434,82 +317,9 @@ func TestCollectionReplica_getPartitionNum(t *testing.T) {
assert.NoError(t, err)
assert.Equal(t, partition.partitionTag, "default")
}
partitionNum, err := (*node.replica).getPartitionNum(UniqueID(0))
assert.NoError(t, err)
assert.Equal(t, partitionNum, 1)
(*node.replica).freeAll()
}
func TestCollectionReplica_addPartition(t *testing.T) {
ctx := context.Background()
node := NewQueryNode(ctx, 0)
collectionName := "collection0"
collectionID := UniqueID(0)
fieldVec := schemapb.FieldSchema{
Name: "vec",
DataType: schemapb.DataType_VECTOR_FLOAT,
TypeParams: []*commonpb.KeyValuePair{
{
Key: "dim",
Value: "16",
},
},
}
fieldInt := schemapb.FieldSchema{
Name: "age",
DataType: schemapb.DataType_INT32,
TypeParams: []*commonpb.KeyValuePair{
{
Key: "dim",
Value: "1",
},
},
}
schema := schemapb.CollectionSchema{
Name: "collection0",
Fields: []*schemapb.FieldSchema{
&fieldVec, &fieldInt,
},
}
collectionMeta := etcdpb.CollectionMeta{
ID: collectionID,
Schema: &schema,
CreateTime: Timestamp(0),
SegmentIDs: []UniqueID{0},
PartitionTags: []string{"default"},
}
collectionMetaBlob := proto.MarshalTextString(&collectionMeta)
assert.NotEqual(t, "", collectionMetaBlob)
var err = (*node.replica).addCollection(&collectionMeta, collectionMetaBlob)
assert.NoError(t, err)
collection, err := (*node.replica).getCollectionByName(collectionName)
assert.NoError(t, err)
assert.Equal(t, collection.meta.Schema.Name, collectionName)
assert.Equal(t, collection.meta.ID, collectionID)
assert.Equal(t, (*node.replica).getCollectionNum(), 1)
for _, tag := range collectionMeta.PartitionTags {
err := (*node.replica).addPartition(collectionID, tag)
assert.NoError(t, err)
partition, err := (*node.replica).getPartitionByTag(collectionID, tag)
assert.NoError(t, err)
assert.Equal(t, partition.partitionTag, "default")
}
(*node.replica).freeAll()
}
func TestCollectionReplica_removePartition(t *testing.T) {
func TestColSegContainer_removePartition(t *testing.T) {
ctx := context.Background()
node := NewQueryNode(ctx, 0)
......@@ -575,157 +385,9 @@ func TestCollectionReplica_removePartition(t *testing.T) {
err = (*node.replica).removePartition(collectionID, partitionTag)
assert.NoError(t, err)
}
(*node.replica).freeAll()
}
func TestCollectionReplica_addPartitionsByCollectionMeta(t *testing.T) {
ctx := context.Background()
node := NewQueryNode(ctx, 0)
collectionName := "collection0"
collectionID := UniqueID(0)
fieldVec := schemapb.FieldSchema{
Name: "vec",
DataType: schemapb.DataType_VECTOR_FLOAT,
TypeParams: []*commonpb.KeyValuePair{
{
Key: "dim",
Value: "16",
},
},
}
fieldInt := schemapb.FieldSchema{
Name: "age",
DataType: schemapb.DataType_INT32,
TypeParams: []*commonpb.KeyValuePair{
{
Key: "dim",
Value: "1",
},
},
}
schema := schemapb.CollectionSchema{
Name: "collection0",
Fields: []*schemapb.FieldSchema{
&fieldVec, &fieldInt,
},
}
collectionMeta := etcdpb.CollectionMeta{
ID: collectionID,
Schema: &schema,
CreateTime: Timestamp(0),
SegmentIDs: []UniqueID{0},
PartitionTags: []string{"p0"},
}
collectionMetaBlob := proto.MarshalTextString(&collectionMeta)
assert.NotEqual(t, "", collectionMetaBlob)
var err = (*node.replica).addCollection(&collectionMeta, collectionMetaBlob)
assert.NoError(t, err)
collection, err := (*node.replica).getCollectionByName(collectionName)
assert.NoError(t, err)
assert.Equal(t, collection.meta.Schema.Name, collectionName)
assert.Equal(t, collection.meta.ID, collectionID)
assert.Equal(t, (*node.replica).getCollectionNum(), 1)
collectionMeta.PartitionTags = []string{"p0", "p1", "p2"}
err = (*node.replica).addPartitionsByCollectionMeta(&collectionMeta)
assert.NoError(t, err)
partitionNum, err := (*node.replica).getPartitionNum(UniqueID(0))
assert.NoError(t, err)
assert.Equal(t, partitionNum, 3)
hasPartition := (*node.replica).hasPartition(UniqueID(0), "p0")
assert.Equal(t, hasPartition, true)
hasPartition = (*node.replica).hasPartition(UniqueID(0), "p1")
assert.Equal(t, hasPartition, true)
hasPartition = (*node.replica).hasPartition(UniqueID(0), "p2")
assert.Equal(t, hasPartition, true)
(*node.replica).freeAll()
}
func TestCollectionReplica_removePartitionsByCollectionMeta(t *testing.T) {
ctx := context.Background()
node := NewQueryNode(ctx, 0)
collectionName := "collection0"
collectionID := UniqueID(0)
fieldVec := schemapb.FieldSchema{
Name: "vec",
DataType: schemapb.DataType_VECTOR_FLOAT,
TypeParams: []*commonpb.KeyValuePair{
{
Key: "dim",
Value: "16",
},
},
}
fieldInt := schemapb.FieldSchema{
Name: "age",
DataType: schemapb.DataType_INT32,
TypeParams: []*commonpb.KeyValuePair{
{
Key: "dim",
Value: "1",
},
},
}
schema := schemapb.CollectionSchema{
Name: "collection0",
Fields: []*schemapb.FieldSchema{
&fieldVec, &fieldInt,
},
}
collectionMeta := etcdpb.CollectionMeta{
ID: collectionID,
Schema: &schema,
CreateTime: Timestamp(0),
SegmentIDs: []UniqueID{0},
PartitionTags: []string{"p0", "p1", "p2"},
}
collectionMetaBlob := proto.MarshalTextString(&collectionMeta)
assert.NotEqual(t, "", collectionMetaBlob)
var err = (*node.replica).addCollection(&collectionMeta, collectionMetaBlob)
assert.NoError(t, err)
collection, err := (*node.replica).getCollectionByName(collectionName)
assert.NoError(t, err)
assert.Equal(t, collection.meta.Schema.Name, collectionName)
assert.Equal(t, collection.meta.ID, collectionID)
assert.Equal(t, (*node.replica).getCollectionNum(), 1)
collectionMeta.PartitionTags = []string{"p0"}
err = (*node.replica).addPartitionsByCollectionMeta(&collectionMeta)
assert.NoError(t, err)
partitionNum, err := (*node.replica).getPartitionNum(UniqueID(0))
assert.NoError(t, err)
assert.Equal(t, partitionNum, 1)
hasPartition := (*node.replica).hasPartition(UniqueID(0), "p0")
assert.Equal(t, hasPartition, true)
hasPartition = (*node.replica).hasPartition(UniqueID(0), "p1")
assert.Equal(t, hasPartition, false)
hasPartition = (*node.replica).hasPartition(UniqueID(0), "p2")
assert.Equal(t, hasPartition, false)
(*node.replica).freeAll()
}
func TestCollectionReplica_getPartitionByTag(t *testing.T) {
func TestColSegContainer_getPartitionByTag(t *testing.T) {
ctx := context.Background()
node := NewQueryNode(ctx, 0)
......@@ -789,78 +451,10 @@ func TestCollectionReplica_getPartitionByTag(t *testing.T) {
assert.Equal(t, partition.partitionTag, "default")
assert.NotNil(t, partition)
}
(*node.replica).freeAll()
}
func TestCollectionReplica_hasPartition(t *testing.T) {
ctx := context.Background()
node := NewQueryNode(ctx, 0)
collectionName := "collection0"
collectionID := UniqueID(0)
fieldVec := schemapb.FieldSchema{
Name: "vec",
DataType: schemapb.DataType_VECTOR_FLOAT,
TypeParams: []*commonpb.KeyValuePair{
{
Key: "dim",
Value: "16",
},
},
}
fieldInt := schemapb.FieldSchema{
Name: "age",
DataType: schemapb.DataType_INT32,
TypeParams: []*commonpb.KeyValuePair{
{
Key: "dim",
Value: "1",
},
},
}
schema := schemapb.CollectionSchema{
Name: "collection0",
Fields: []*schemapb.FieldSchema{
&fieldVec, &fieldInt,
},
}
collectionMeta := etcdpb.CollectionMeta{
ID: collectionID,
Schema: &schema,
CreateTime: Timestamp(0),
SegmentIDs: []UniqueID{0},
PartitionTags: []string{"default"},
}
collectionMetaBlob := proto.MarshalTextString(&collectionMeta)
assert.NotEqual(t, "", collectionMetaBlob)
var err = (*node.replica).addCollection(&collectionMeta, collectionMetaBlob)
assert.NoError(t, err)
collection, err := (*node.replica).getCollectionByName(collectionName)
assert.NoError(t, err)
assert.Equal(t, collection.meta.Schema.Name, collectionName)
assert.Equal(t, collection.meta.ID, collectionID)
assert.Equal(t, (*node.replica).getCollectionNum(), 1)
err = (*node.replica).addPartition(collectionID, collectionMeta.PartitionTags[0])
assert.NoError(t, err)
hasPartition := (*node.replica).hasPartition(UniqueID(0), "default")
assert.Equal(t, hasPartition, true)
hasPartition = (*node.replica).hasPartition(UniqueID(0), "default1")
assert.Equal(t, hasPartition, false)
(*node.replica).freeAll()
}
//----------------------------------------------------------------------------------------------------- segment
func TestCollectionReplica_addSegment(t *testing.T) {
func TestColSegContainer_addSegment(t *testing.T) {
ctx := context.Background()
node := NewQueryNode(ctx, 0)
......@@ -927,11 +521,9 @@ func TestCollectionReplica_addSegment(t *testing.T) {
assert.NoError(t, err)
assert.Equal(t, targetSeg.segmentID, UniqueID(i))
}
(*node.replica).freeAll()
}
func TestCollectionReplica_removeSegment(t *testing.T) {
func TestColSegContainer_removeSegment(t *testing.T) {
ctx := context.Background()
node := NewQueryNode(ctx, 0)
......@@ -1000,11 +592,9 @@ func TestCollectionReplica_removeSegment(t *testing.T) {
err = (*node.replica).removeSegment(UniqueID(i))
assert.NoError(t, err)
}
(*node.replica).freeAll()
}
func TestCollectionReplica_getSegmentByID(t *testing.T) {
func TestColSegContainer_getSegmentByID(t *testing.T) {
ctx := context.Background()
node := NewQueryNode(ctx, 0)
......@@ -1071,11 +661,9 @@ func TestCollectionReplica_getSegmentByID(t *testing.T) {
assert.NoError(t, err)
assert.Equal(t, targetSeg.segmentID, UniqueID(i))
}
(*node.replica).freeAll()
}
func TestCollectionReplica_hasSegment(t *testing.T) {
func TestColSegContainer_hasSegment(t *testing.T) {
ctx := context.Background()
node := NewQueryNode(ctx, 0)
......@@ -1146,81 +734,4 @@ func TestCollectionReplica_hasSegment(t *testing.T) {
hasSeg = (*node.replica).hasSegment(UniqueID(i + 100))
assert.Equal(t, hasSeg, false)
}
(*node.replica).freeAll()
}
func TestCollectionReplica_freeAll(t *testing.T) {
ctx := context.Background()
node := NewQueryNode(ctx, 0)
collectionName := "collection0"
collectionID := UniqueID(0)
fieldVec := schemapb.FieldSchema{
Name: "vec",
DataType: schemapb.DataType_VECTOR_FLOAT,
TypeParams: []*commonpb.KeyValuePair{
{
Key: "dim",
Value: "16",
},
},
}
fieldInt := schemapb.FieldSchema{
Name: "age",
DataType: schemapb.DataType_INT32,
TypeParams: []*commonpb.KeyValuePair{
{
Key: "dim",
Value: "1",
},
},
}
schema := schemapb.CollectionSchema{
Name: "collection0",
Fields: []*schemapb.FieldSchema{
&fieldVec, &fieldInt,
},
}
collectionMeta := etcdpb.CollectionMeta{
ID: collectionID,
Schema: &schema,
CreateTime: Timestamp(0),
SegmentIDs: []UniqueID{0},
PartitionTags: []string{"default"},
}
collectionMetaBlob := proto.MarshalTextString(&collectionMeta)
assert.NotEqual(t, "", collectionMetaBlob)
var err = (*node.replica).addCollection(&collectionMeta, collectionMetaBlob)
assert.NoError(t, err)
collection, err := (*node.replica).getCollectionByName(collectionName)
assert.NoError(t, err)
assert.Equal(t, collection.meta.Schema.Name, collectionName)
assert.Equal(t, collection.meta.ID, UniqueID(0))
assert.Equal(t, (*node.replica).getCollectionNum(), 1)
err = (*node.replica).addPartition(collectionID, collectionMeta.PartitionTags[0])
assert.NoError(t, err)
const segmentNum = 3
for i := 0; i < segmentNum; i++ {
err := (*node.replica).addSegment(UniqueID(i), collectionMeta.PartitionTags[0], collectionID)
assert.NoError(t, err)
targetSeg, err := (*node.replica).getSegmentByID(UniqueID(i))
assert.NoError(t, err)
assert.Equal(t, targetSeg.segmentID, UniqueID(i))
hasSeg := (*node.replica).hasSegment(UniqueID(i))
assert.Equal(t, hasSeg, true)
hasSeg = (*node.replica).hasSegment(UniqueID(i + 100))
assert.Equal(t, hasSeg, false)
}
(*node.replica).freeAll()
}
internal/reader/meta_service.go
浏览文件 @ d1f31c7b
......@@ -214,18 +214,6 @@ func (mService *metaService) processSegmentModify(id string, value string) {
func (mService *metaService) processCollectionModify(id string, value string) {
println("Modify Collection: ", id)
col := mService.collectionUnmarshal(value)
if col != nil {
err := (*mService.replica).addPartitionsByCollectionMeta(col)
if err != nil {
log.Println(err)
}
err = (*mService.replica).removePartitionsByCollectionMeta(col)
if err != nil {
log.Println(err)
}
}
}
func (mService *metaService) processModify(key string, msg string) {
......
internal/reader/meta_service_test.go
浏览文件 @ d1f31c7b
......@@ -452,9 +452,7 @@ func TestMetaService_processCollectionModify(t *testing.T) {
>
>
segmentIDs: 0
partition_tags: "p0"
partition_tags: "p1"
partition_tags: "p2"
partition_tags: "default"
`
(*node.metaService).processCollectionCreate(id, value)
......@@ -465,19 +463,7 @@ func TestMetaService_processCollectionModify(t *testing.T) {
assert.NoError(t, err)
assert.Equal(t, collection.ID(), UniqueID(0))
partitionNum, err := (*node.replica).getPartitionNum(UniqueID(0))
assert.NoError(t, err)
assert.Equal(t, partitionNum, 3)
hasPartition := (*node.replica).hasPartition(UniqueID(0), "p0")
assert.Equal(t, hasPartition, true)
hasPartition = (*node.replica).hasPartition(UniqueID(0), "p1")
assert.Equal(t, hasPartition, true)
hasPartition = (*node.replica).hasPartition(UniqueID(0), "p2")
assert.Equal(t, hasPartition, true)
hasPartition = (*node.replica).hasPartition(UniqueID(0), "p3")
assert.Equal(t, hasPartition, false)
// TODO: use different index for testing processCollectionModify
newValue := `schema: <
name: "test"
fields: <
......@@ -498,28 +484,13 @@ func TestMetaService_processCollectionModify(t *testing.T) {
>
>
segmentIDs: 0
partition_tags: "p1"
partition_tags: "p2"
partition_tags: "p3"
partition_tags: "default"
`
(*node.metaService).processCollectionModify(id, newValue)
collection, err = (*node.replica).getCollectionByName("test")
assert.NoError(t, err)
assert.Equal(t, collection.ID(), UniqueID(0))
partitionNum, err = (*node.replica).getPartitionNum(UniqueID(0))
assert.NoError(t, err)
assert.Equal(t, partitionNum, 3)
hasPartition = (*node.replica).hasPartition(UniqueID(0), "p0")
assert.Equal(t, hasPartition, false)
hasPartition = (*node.replica).hasPartition(UniqueID(0), "p1")
assert.Equal(t, hasPartition, true)
hasPartition = (*node.replica).hasPartition(UniqueID(0), "p2")
assert.Equal(t, hasPartition, true)
hasPartition = (*node.replica).hasPartition(UniqueID(0), "p3")
assert.Equal(t, hasPartition, true)
}
func TestMetaService_processModify(t *testing.T) {
......@@ -552,9 +523,7 @@ func TestMetaService_processModify(t *testing.T) {
>
>
segmentIDs: 0
partition_tags: "p0"
partition_tags: "p1"
partition_tags: "p2"
partition_tags: "default"
`
(*node.metaService).processCreate(key1, msg1)
......@@ -565,21 +534,8 @@ func TestMetaService_processModify(t *testing.T) {
assert.NoError(t, err)
assert.Equal(t, collection.ID(), UniqueID(0))
partitionNum, err := (*node.replica).getPartitionNum(UniqueID(0))
assert.NoError(t, err)
assert.Equal(t, partitionNum, 3)
hasPartition := (*node.replica).hasPartition(UniqueID(0), "p0")
assert.Equal(t, hasPartition, true)
hasPartition = (*node.replica).hasPartition(UniqueID(0), "p1")
assert.Equal(t, hasPartition, true)
hasPartition = (*node.replica).hasPartition(UniqueID(0), "p2")
assert.Equal(t, hasPartition, true)
hasPartition = (*node.replica).hasPartition(UniqueID(0), "p3")
assert.Equal(t, hasPartition, false)
key2 := "by-dev/segment/0"
msg2 := `partition_tag: "p1"
msg2 := `partition_tag: "default"
channel_start: 0
channel_end: 128
close_time: 18446744073709551615
......@@ -612,9 +568,7 @@ func TestMetaService_processModify(t *testing.T) {
>
>
segmentIDs: 0
partition_tags: "p1"
partition_tags: "p2"
partition_tags: "p3"
partition_tags: "default"
`
(*node.metaService).processModify(key1, msg3)
......@@ -622,25 +576,13 @@ func TestMetaService_processModify(t *testing.T) {
assert.NoError(t, err)
assert.Equal(t, collection.ID(), UniqueID(0))
partitionNum, err = (*node.replica).getPartitionNum(UniqueID(0))
assert.NoError(t, err)
assert.Equal(t, partitionNum, 3)
hasPartition = (*node.replica).hasPartition(UniqueID(0), "p0")
assert.Equal(t, hasPartition, false)
hasPartition = (*node.replica).hasPartition(UniqueID(0), "p1")
assert.Equal(t, hasPartition, true)
hasPartition = (*node.replica).hasPartition(UniqueID(0), "p2")
assert.Equal(t, hasPartition, true)
hasPartition = (*node.replica).hasPartition(UniqueID(0), "p3")
assert.Equal(t, hasPartition, true)
msg4 := `partition_tag: "p1"
msg4 := `partition_tag: "default"
channel_start: 0
channel_end: 128
close_time: 18446744073709551615
`
// TODO: modify segment for testing processCollectionModify
(*node.metaService).processModify(key2, msg4)
seg, err := (*node.replica).getSegmentByID(UniqueID(0))
assert.NoError(t, err)
......
Markdown is supported
0% 或 .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑！
想要评论请 注册 或