Merge pull request #101 from zilliztech/feature_cmd_support_collection

Feature support collection Disable <-> Enable AutoId migration

README_2X.md

@@ -82,6 +82,11 @@ meta:
     closeDynamicField: false      # If not, the source collection DynamicField prop will be used.
     shardNum: 2                   # If not, the source collection ShardNum prop will be used.
     consistencyLevel: Customized  # If not, the source collection consistencyLevel prop will be used.
+    #about autoId: default "", when want migration collection between Disable and Enable Auto property is very useful.
+    #if "" mean the source collection AutoId prop will be used. 
+    #if "true" mean enable AutoId, 
+    #if "false" mean disable AutoId,
+    autoId: "" 
   #......  
 ...
 ```

core/config/resolve_meta.go

@@ -91,6 +91,10 @@ func resolveMilvusCfg(v *viper.Viper) *milvustype.MilvusCfg {
 			if ok {
 				milvus.Collection = collName
 			}
+			autoId, ok := milvusMap["autoid"].(string)
+			if ok {
+				milvus.AutoId = autoId
+			}
 			shardNum, ok := milvusMap["shardnum"].(int)
 			if ok {
 				milvus.ShardNum = shardNum

core/dbclient/cus_field_milvus2x.go

@@ -51,7 +51,12 @@ func (cus *CustomFieldMilvus2x) hasCollection(ctx context.Context, collection st
 }
 
 func (cus *CustomFieldMilvus2x) createCollection(ctx context.Context, collectionInfo *common.CollectionInfo) error {
-	log.Info("Begin to Create Custom field Milvus Collection,", zap.String("collection", collectionInfo.Param.CollectionName))
+	log.Info("Begin to Create Custom field Milvus Collection,",
+		zap.String("collection", collectionInfo.Param.CollectionName),
+		zap.Any("fields", collectionInfo.Fields),
+		zap.Bool("dynamicField", collectionInfo.Param.EnableDynamicField),
+		zap.Bool("autoId", collectionInfo.Param.AutoId),
+		zap.String("description", collectionInfo.Param.Description))
 	// schema
 	schema := &entity.Schema{
 		CollectionName:     collectionInfo.Param.CollectionName,

core/reader/source/milvus2x_source.go

@@ -57,7 +57,7 @@ func (milvus2xSource *Milvus2xSource) ReadFirst(ctx context.Context) (*milvus2x.
 }
 
 func (milvus2xSource *Milvus2xSource) removePKColIfOpenAutoId(data *milvus2x.Milvus2xData) {
-	if milvus2xSource.CollCfg.MilvusCfg.AutoId {
+	if milvus2xSource.CollCfg.MilvusCfg.AutoId == "true" {
 		for idx, dataColumn := range data.Columns {
 			if dataColumn.Name() == milvus2xSource.CollCfg.MilvusCfg.PkName {
 				delPkColList := append(data.Columns[:idx], data.Columns[idx+1:]...)

core/transform/milvus2x/convert/milvus2x_convert.go

@@ -33,17 +33,23 @@ func ToMilvusParam(ctx context.Context, collCfg *milvus2xtype.CollectionCfg, mil
 	if Description == "" {
 		Description = "Migration from Milvus2x"
 	}
+
 	//collCfg.MilvusCfg.AutoId = srcCollEntity.Schema.AutoID
 	//log.Info("milvus2x transform to custom Milvus", zap.Any("milvusCfg AutoId", collCfg.MilvusCfg.AutoId))
 	//log.Info("milvus2x transform to custom Milvus", zap.Any("srcColl AutoId", srcCollEntity.Schema.AutoID))
 	param := &common.CollectionParam{
 		CollectionName:     ToMilvusCollectionName(collCfg),
 		ShardsNum:          ToShardNum(collCfg.MilvusCfg.ShardNum, srcCollEntity),
 		EnableDynamicField: !collCfg.MilvusCfg.CloseDynamicField,
-		AutoId:             collCfg.MilvusCfg.AutoId,
+		//AutoId:             collCfg.MilvusCfg.AutoId,
 		//Description:        "Migration from Milvus2x",
 		Description: Description,
 	}
+	if collCfg.MilvusCfg.AutoId == "true" {
+		param.AutoId = true
+	} else {
+		param.AutoId = false
+	}
 	param.ConsistencyLevel, err = GetMilvusConsistencyLevel(collCfg, srcCollEntity)
 	if err != nil {
 		return nil, err
@@ -80,7 +86,7 @@ func fillAllFileds(collEntity *entity.Collection, collCfg *milvus2xtype.Collecti
 		queryFields = append(queryFields, cfgField)
 		if srcField.PrimaryKey {
 			log.Info("milvus2x transform to fillAllFields Milvus", zap.Any("srcField AutoId", srcField.AutoID))
-			collCfg.MilvusCfg.AutoId = srcField.AutoID
+			setTargetCollAutoIdProperty(collCfg, srcField)
 			collCfg.MilvusCfg.PkName = srcField.Name
 		}
 	}
@@ -91,18 +97,18 @@ func fillAllFileds(collEntity *entity.Collection, collCfg *milvus2xtype.Collecti
 func fillCustomFileds(collEntity *entity.Collection, collCfg *milvus2xtype.CollectionCfg) ([]*entity.Field, error) {
 	var _fields []*entity.Field
 
-	var existPKField = false
+	//var existPKField = false
 	var existVectorField = false
 	for _, field := range collCfg.Fields {
 		var matchField *entity.Field
 		for _, srcField := range collEntity.Schema.Fields {
 			if srcField.Name == field.Name {
 				matchField = srcField
 				if srcField.PrimaryKey {
-					existPKField = true
+					//existPKField = true
 					field.PK = true
 					log.Info("milvus2x transform to fillCustomFields Milvus", zap.Any("srcField AutoId", srcField.AutoID))
-					collCfg.MilvusCfg.AutoId = srcField.AutoID
+					setTargetCollAutoIdProperty(collCfg, srcField)
 					collCfg.MilvusCfg.PkName = srcField.Name
 				}
 				if convert.IsVectorField(srcField) {
@@ -115,15 +121,33 @@ func fillCustomFileds(collEntity *entity.Collection, collCfg *milvus2xtype.Colle
 		}
 		_fields = append(_fields, matchField)
 	}
-	if existPKField == false {
-		return nil, errors.New("not migrate milvus2x source collection PrimaryKey field")
-	}
+	//if existPKField == false {
+	//	return nil, errors.New("not migrate milvus2x source collection PrimaryKey field")
+	//}
 	if existVectorField == false {
 		return nil, errors.New("not migrate milvus2x source collection FloatVector type field")
 	}
 	return _fields, nil
 }
 
+func setTargetCollAutoIdProperty(collCfg *milvus2xtype.CollectionCfg, srcField *entity.Field) {
+	log.Info("milvus2x transform custom target Milvus", zap.Any("AutoId", collCfg.MilvusCfg.AutoId))
+	//如果用户没有设置target表AutoId属性，则copy source表的AutoId属性, (主要给后面是否要迁移ID字段判断使用)
+	if collCfg.MilvusCfg.AutoId == "" {
+		if srcField.AutoID {
+			collCfg.MilvusCfg.AutoId = "true"
+		} else {
+			collCfg.MilvusCfg.AutoId = "false"
+		}
+	}
+	//创建target时，已字段上的autoId来判断是否开启AutoId （创建表时使用）
+	if collCfg.MilvusCfg.AutoId == "true" {
+		srcField.AutoID = true
+	} else {
+		srcField.AutoID = false
+	}
+}
+
 func ToMilvusCollectionName(collCfg *milvus2xtype.CollectionCfg) string {
 	if len(collCfg.MilvusCfg.Collection) > 0 {
 		return collCfg.MilvusCfg.Collection

core/type/milvustype/type.go

@@ -9,7 +9,7 @@ type MilvusCfg struct {
 	ConsistencyLevel  string `json:"consistencyLevel"`  //default value: ""
 	LoadData          bool   `json:"loadData"`          //default value: false
 	CreateIndex       bool   `json:"createIndex"`       //default value: false
-	AutoId            bool   `json:"autoId"`
+	AutoId            string `json:"autoId"`            //"": copy source collection, "true":enableAutoId, "false":"closeAutoId",
 	PkName            string `json:"pkName"`
 }
 

go.mod

@@ -14,7 +14,7 @@ require (
 	github.com/lingdor/stackerror v0.0.0-20191119040541-976d8885ed76
 	github.com/matoous/go-nanoid/v2 v2.0.0
 	github.com/milvus-io/milvus-sdk-go v1.1.1
-	github.com/milvus-io/milvus-sdk-go/v2 v2.4.2-0.20240819094545-542b5a0158c1
+	github.com/milvus-io/milvus-sdk-go/v2 v2.4.2-0.20240821073850-0c84a71df153
 	github.com/minio/minio-go/v7 v7.0.66
 	github.com/olivere/elastic/v7 v7.0.32
 	github.com/orcaman/concurrent-map/v2 v2.0.1
@@ -33,7 +33,7 @@ require (
 	go.uber.org/atomic v1.11.0
 	go.uber.org/zap v1.26.0
 	golang.org/x/oauth2 v0.15.0
-	golang.org/x/sync v0.5.0
+	golang.org/x/sync v0.8.0
 	golang.org/x/time v0.5.0
 	google.golang.org/grpc v1.60.1
 	gopkg.in/natefinch/lumberjack.v2 v2.2.1
@@ -92,7 +92,7 @@ require (
 	github.com/mailru/easyjson v0.7.7 // indirect
 	github.com/mattn/go-isatty v0.0.20 // indirect
 	github.com/mattn/go-sqlite3 v1.14.19 // indirect
-	github.com/milvus-io/milvus-proto/go-api/v2 v2.4.6 // indirect
+	github.com/milvus-io/milvus-proto/go-api/v2 v2.4.10-0.20240819025435-512e3b98866a // indirect
 	github.com/minio/md5-simd v1.1.2 // indirect
 	github.com/minio/sha256-simd v1.0.1 // indirect
 	github.com/mitchellh/mapstructure v1.5.0 // indirect

go.sum

@@ -187,10 +187,14 @@ github.com/mattn/go-sqlite3 v1.14.19 h1:fhGleo2h1p8tVChob4I9HpmVFIAkKGpiukdrgQbW
 github.com/mattn/go-sqlite3 v1.14.19/go.mod h1:2eHXhiwb8IkHr+BDWZGa96P6+rkvnG63S2DGjv9HUNg=
 github.com/milvus-io/milvus-proto/go-api/v2 v2.4.6 h1:rSkwp5Mg/7KBSUqXcrPBUgTQGZNdvYWEKB+rHo9YJtk=
 github.com/milvus-io/milvus-proto/go-api/v2 v2.4.6/go.mod h1:1OIl0v5PQeNxIJhCvY+K55CBUOYDZevw9g9380u1Wek=
+github.com/milvus-io/milvus-proto/go-api/v2 v2.4.10-0.20240819025435-512e3b98866a h1:0B/8Fo66D8Aa23Il0yrQvg1KKz92tE/BJ5BvkUxxAAk=
+github.com/milvus-io/milvus-proto/go-api/v2 v2.4.10-0.20240819025435-512e3b98866a/go.mod h1:1OIl0v5PQeNxIJhCvY+K55CBUOYDZevw9g9380u1Wek=
 github.com/milvus-io/milvus-sdk-go v1.1.1 h1:QseeGBb92T4ny5jSscaWJ+toG74p5zE0kaxFyeskWSE=
 github.com/milvus-io/milvus-sdk-go v1.1.1/go.mod h1:jyYc5vzEQS9V+ZrSL2h/oeKcYiixvKBV/TrxXbLu01w=
 github.com/milvus-io/milvus-sdk-go/v2 v2.4.2-0.20240819094545-542b5a0158c1 h1:tmPS7djDZ0VrXmgp/UbWKDFwsBuIVTjF+p4jF4Pd3KI=
 github.com/milvus-io/milvus-sdk-go/v2 v2.4.2-0.20240819094545-542b5a0158c1/go.mod h1:kQ7SuDWugDhCqEHIvokitjVO9CyATIFa+h1dHEk86Zc=
+github.com/milvus-io/milvus-sdk-go/v2 v2.4.2-0.20240821073850-0c84a71df153 h1:3kIbxhjuXuVUMOnJI3R1oPkfM4IIRKs7W1tH1spnLnk=
+github.com/milvus-io/milvus-sdk-go/v2 v2.4.2-0.20240821073850-0c84a71df153/go.mod h1:ulO1YUXKH0PGg50q27grw048GDY9ayB4FPmh7D+FFTA=
 github.com/minio/md5-simd v1.1.2 h1:Gdi1DZK69+ZVMoNHRXJyNcxrMA4dSxoYHZSQbirFg34=
 github.com/minio/md5-simd v1.1.2/go.mod h1:MzdKDxYpY2BT9XQFocsiZf/NKVtR7nkE4RoEpN+20RM=
 github.com/minio/minio-go/v7 v7.0.66 h1:bnTOXOHjOqv/gcMuiVbN9o2ngRItvqE774dG9nq0Dzw=
@@ -366,6 +370,8 @@ golang.org/x/sync v0.0.0-20220722155255-886fb9371eb4/go.mod h1:RxMgew5VJxzue5/jJ
 golang.org/x/sync v0.1.0/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.5.0 h1:60k92dhOjHxJkrqnwsfl8KuaHbn/5dl0lUPUklKo3qE=
 golang.org/x/sync v0.5.0/go.mod h1:Czt+wKu1gCyEFDUtn0jG5QVvpJ6rzVqr5aXyt9drQfk=
+golang.org/x/sync v0.8.0 h1:3NFvSEYkUoMifnESzZl15y791HH1qU2xm6eCJU5ZPXQ=
+golang.org/x/sync v0.8.0/go.mod h1:Czt+wKu1gCyEFDUtn0jG5QVvpJ6rzVqr5aXyt9drQfk=
 golang.org/x/sys v0.0.0-20180830151530-49385e6e1522/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20190412213103-97732733099d/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=

storage/milvus2x/milvus2_3_ver.go

@@ -46,7 +46,7 @@ func (milvus23 *Milvus23VerClient) InitIterator(ctx context.Context, collCfg *mi
 		zap.Int("BatchSize", batchSize), zap.Int64("CollectionRow", count))
 	fieldNames := make([]string, 0, len(collCfg.Fields))
 	for _, fieldCfg := range collCfg.Fields {
-		if collCfg.MilvusCfg.AutoId && fieldCfg.PK {
+		if collCfg.MilvusCfg.AutoId == "true" && fieldCfg.PK {
 			continue
 		}
 		fieldNames = append(fieldNames, fieldCfg.Name)