Merge branch 'master' of https://github.com/milvus-io/milvus into 241…

…2-opt-meta-mutex

README.md

@@ -250,7 +250,7 @@ Reference to cite when you use Milvus in a research paper:
 }
 ```
 <!-- Do not remove start of hero-bot -->
-<img src="https://img.shields.io/badge/all--contributors-420-orange"><br>
+<img src="https://img.shields.io/badge/all--contributors-421-orange"><br>
 <a href="https://github.com/0xflotus"><img src="https://avatars.githubusercontent.com/u/26602940?v=4" width="30px" /></a>
 <a href="https://github.com/ABNER-1"><img src="https://avatars.githubusercontent.com/u/24547351?v=4" width="30px" /></a>
 <a href="https://github.com/Accagain2014"><img src="https://avatars.githubusercontent.com/u/9635216?v=4" width="30px" /></a>
@@ -385,6 +385,7 @@ Reference to cite when you use Milvus in a research paper:
 <a href="https://github.com/bjzhjing"><img src="https://avatars.githubusercontent.com/u/46661806?v=4" width="30px" /></a>
 <a href="https://github.com/bo-huang"><img src="https://avatars.githubusercontent.com/u/24309515?v=4" width="30px" /></a>
 <a href="https://github.com/brandonbiggs"><img src="https://avatars.githubusercontent.com/u/34954680?v=4" width="30px" /></a>
+<a href="https://github.com/brcarry"><img src="https://avatars.githubusercontent.com/u/58139344?v=4" width="30px" /></a>
 <a href="https://github.com/brunocfnba"><img src="https://avatars.githubusercontent.com/u/7377163?v=4" width="30px" /></a>
 <a href="https://github.com/bryanwux"><img src="https://avatars.githubusercontent.com/u/17968665?v=4" width="30px" /></a>
 <a href="https://github.com/caesarjuly"><img src="https://avatars.githubusercontent.com/u/927521?v=4" width="30px" /></a>

README_CN.md

@@ -183,7 +183,7 @@ Milvus [训练营](https://github.com/milvus-io/bootcamp)能够帮助你了解
 ### All contributors
 
 <br><!-- Do not remove start of hero-bot -->
-<img src="https://img.shields.io/badge/all--contributors-420-orange"><br>
+<img src="https://img.shields.io/badge/all--contributors-421-orange"><br>
 <a href="https://github.com/0xflotus"><img src="https://avatars.githubusercontent.com/u/26602940?v=4" width="30px" /></a>
 <a href="https://github.com/ABNER-1"><img src="https://avatars.githubusercontent.com/u/24547351?v=4" width="30px" /></a>
 <a href="https://github.com/Accagain2014"><img src="https://avatars.githubusercontent.com/u/9635216?v=4" width="30px" /></a>
@@ -318,6 +318,7 @@ Milvus [训练营](https://github.com/milvus-io/bootcamp)能够帮助你了解
 <a href="https://github.com/bjzhjing"><img src="https://avatars.githubusercontent.com/u/46661806?v=4" width="30px" /></a>
 <a href="https://github.com/bo-huang"><img src="https://avatars.githubusercontent.com/u/24309515?v=4" width="30px" /></a>
 <a href="https://github.com/brandonbiggs"><img src="https://avatars.githubusercontent.com/u/34954680?v=4" width="30px" /></a>
+<a href="https://github.com/brcarry"><img src="https://avatars.githubusercontent.com/u/58139344?v=4" width="30px" /></a>
 <a href="https://github.com/brunocfnba"><img src="https://avatars.githubusercontent.com/u/7377163?v=4" width="30px" /></a>
 <a href="https://github.com/bryanwux"><img src="https://avatars.githubusercontent.com/u/17968665?v=4" width="30px" /></a>
 <a href="https://github.com/caesarjuly"><img src="https://avatars.githubusercontent.com/u/927521?v=4" width="30px" /></a>

cmd/components/data_coord.go

@@ -58,10 +58,10 @@ func (s *DataCoord) Prepare() error {
 // Run starts service
 func (s *DataCoord) Run() error {
 	if err := s.svr.Run(); err != nil {
-		log.Error("DataCoord starts error", zap.Error(err))
+		log.Ctx(s.ctx).Error("DataCoord starts error", zap.Error(err))
 		return err
 	}
-	log.Debug("DataCoord successfully started")
+	log.Ctx(s.ctx).Debug("DataCoord successfully started")
 	return nil
 }
 

cmd/components/data_node.go

@@ -57,10 +57,10 @@ func (d *DataNode) Prepare() error {
 // Run starts service
 func (d *DataNode) Run() error {
 	if err := d.svr.Run(); err != nil {
-		log.Error("DataNode starts error", zap.Error(err))
+		log.Ctx(d.ctx).Error("DataNode starts error", zap.Error(err))
 		return err
 	}
-	log.Debug("Datanode successfully started")
+	log.Ctx(d.ctx).Info("Datanode successfully started")
 	return nil
 }
 

cmd/components/index_coord.go

@@ -39,13 +39,13 @@ func (s *IndexCoord) Prepare() error {
 
 // Run starts service
 func (s *IndexCoord) Run() error {
-	log.Info("IndexCoord running ...")
+	log.Ctx(context.TODO()).Info("IndexCoord running ...")
 	return nil
 }
 
 // Stop terminates service
 func (s *IndexCoord) Stop() error {
-	log.Info("IndexCoord stopping ...")
+	log.Ctx(context.TODO()).Info("IndexCoord stopping ...")
 	return nil
 }
 

cmd/components/index_node.go

@@ -33,13 +33,16 @@ import (
 
 // IndexNode implements IndexNode grpc server
 type IndexNode struct {
+	ctx context.Context
 	svr *grpcindexnode.Server
 }
 
 // NewIndexNode creates a new IndexNode
 func NewIndexNode(ctx context.Context, factory dependency.Factory) (*IndexNode, error) {
 	var err error
-	n := &IndexNode{}
+	n := &IndexNode{
+		ctx: ctx,
+	}
 	svr, err := grpcindexnode.NewServer(ctx, factory)
 	if err != nil {
 		return nil, err
@@ -55,10 +58,10 @@ func (n *IndexNode) Prepare() error {
 // Run starts service
 func (n *IndexNode) Run() error {
 	if err := n.svr.Run(); err != nil {
-		log.Error("IndexNode starts error", zap.Error(err))
+		log.Ctx(n.ctx).Error("IndexNode starts error", zap.Error(err))
 		return err
 	}
-	log.Debug("IndexNode successfully started")
+	log.Ctx(n.ctx).Info("IndexNode successfully started")
 	return nil
 }
 

cmd/components/proxy.go

@@ -58,10 +58,10 @@ func (n *Proxy) Prepare() error {
 // Run starts service
 func (n *Proxy) Run() error {
 	if err := n.svr.Run(); err != nil {
-		log.Error("Proxy starts error", zap.Error(err))
+		log.Ctx(context.TODO()).Error("Proxy starts error", zap.Error(err))
 		return err
 	}
-	log.Info("Proxy successfully started")
+	log.Ctx(context.TODO()).Info("Proxy successfully started")
 	return nil
 }
 

cmd/components/query_coord.go

@@ -57,10 +57,10 @@ func (qs *QueryCoord) Prepare() error {
 // Run starts service
 func (qs *QueryCoord) Run() error {
 	if err := qs.svr.Run(); err != nil {
-		log.Error("QueryCoord starts error", zap.Error(err))
+		log.Ctx(qs.ctx).Error("QueryCoord starts error", zap.Error(err))
 		return err
 	}
-	log.Debug("QueryCoord successfully started")
+	log.Ctx(qs.ctx).Info("QueryCoord successfully started")
 	return nil
 }
 

cmd/components/query_node.go

@@ -57,10 +57,10 @@ func (q *QueryNode) Prepare() error {
 // Run starts service
 func (q *QueryNode) Run() error {
 	if err := q.svr.Run(); err != nil {
-		log.Error("QueryNode starts error", zap.Error(err))
+		log.Ctx(q.ctx).Error("QueryNode starts error", zap.Error(err))
 		return err
 	}
-	log.Debug("QueryNode successfully started")
+	log.Ctx(q.ctx).Info("QueryNode successfully started")
 	return nil
 }
 

cmd/components/root_coord.go

@@ -56,10 +56,10 @@ func (rc *RootCoord) Prepare() error {
 // Run starts service
 func (rc *RootCoord) Run() error {
 	if err := rc.svr.Run(); err != nil {
-		log.Error("RootCoord starts error", zap.Error(err))
+		log.Ctx(rc.ctx).Error("RootCoord starts error", zap.Error(err))
 		return err
 	}
-	log.Info("RootCoord successfully started")
+	log.Ctx(rc.ctx).Info("RootCoord successfully started")
 	return nil
 }
 

cmd/components/streaming_node.go

@@ -29,8 +29,8 @@ type StreamingNode struct {
 }
 
 // NewStreamingNode creates a new StreamingNode
-func NewStreamingNode(_ context.Context, factory dependency.Factory) (*StreamingNode, error) {
-	svr, err := streamingnode.NewServer(factory)
+func NewStreamingNode(ctx context.Context, factory dependency.Factory) (*StreamingNode, error) {
+	svr, err := streamingnode.NewServer(ctx, factory)
 	if err != nil {
 		return nil, err
 	}

cmd/milvus/mck.go

@@ -106,6 +106,7 @@ func (c *mck) execute(args []string, flags *flag.FlagSet) {
 
 func (c *mck) run() {
 	c.connectMinio()
+	log := log.Ctx(context.TODO())
 
 	_, values, err := c.metaKV.LoadWithPrefix(context.TODO(), segmentPrefix)
 	if err != nil {
@@ -206,13 +207,14 @@ func (c *mck) formatFlags(args []string, flags *flag.FlagSet) {
 	if err := flags.Parse(os.Args[2:]); err != nil {
 		log.Fatal("failed to parse flags", zap.Error(err))
 	}
-	log.Info("args", zap.Strings("args", args))
+	log.Ctx(context.TODO()).Info("args", zap.Strings("args", args))
 }
 
 func (c *mck) connectEctd() {
 	c.params.Init(paramtable.NewBaseTable())
 	var etcdCli *clientv3.Client
 	var err error
+	log := log.Ctx(context.TODO())
 	if c.etcdIP != "" {
 		etcdCli, err = etcd.GetRemoteEtcdClient([]string{c.etcdIP})
 	} else {
@@ -243,7 +245,7 @@ func (c *mck) connectMinio() {
 	var err error
 	c.minioChunkManager, err = chunkManagerFactory.NewPersistentStorageChunkManager(context.Background())
 	if err != nil {
-		log.Fatal("failed to connect to minio", zap.Error(err))
+		log.Ctx(context.TODO()).Fatal("failed to connect to minio", zap.Error(err))
 	}
 }
 
@@ -254,11 +256,12 @@ func getConfigValue(a string, b string, name string) string {
 	if b != "" {
 		return b
 	}
-	log.Panic(fmt.Sprintf("the config '%s' is empty", name))
+	log.Ctx(context.TODO()).Panic(fmt.Sprintf("the config '%s' is empty", name))
 	return ""
 }
 
 func (c *mck) cleanTrash() {
+	log := log.Ctx(context.TODO())
 	keys, _, err := c.metaKV.LoadWithPrefix(context.TODO(), MckTrash)
 	if err != nil {
 		log.Error("failed to load backup info", zap.Error(err))
@@ -367,6 +370,7 @@ func getTrashKey(taskType, key string) string {
 }
 
 func (c *mck) extractTask(prefix string, keys []string, values []string) {
+	log := log.Ctx(context.TODO())
 	for i := range keys {
 		taskID, err := strconv.ParseInt(filepath.Base(keys[i]), 10, 64)
 		if err != nil {
@@ -393,6 +397,7 @@ func (c *mck) extractTask(prefix string, keys []string, values []string) {
 }
 
 func (c *mck) removeTask(invalidTask int64) bool {
+	log := log.Ctx(context.TODO())
 	taskType := c.taskNameMap[invalidTask]
 	key := c.taskKeyMap[invalidTask]
 	err := c.metaKV.Save(context.TODO(), getTrashKey(taskType, key), c.allTaskInfo[key])
@@ -520,6 +525,7 @@ func (c *mck) extractVecFieldIndexInfo(taskID int64, infos []*querypb.FieldIndex
 
 // return partitionIDs,segmentIDs,error
 func (c *mck) unmarshalTask(taskID int64, t string) (string, []int64, []int64, error) {
+	log := log.Ctx(context.TODO())
 	header := commonpb.MsgHeader{}
 	err := proto.Unmarshal([]byte(t), &header)
 	if err != nil {

cmd/milvus/util.go

@@ -242,7 +242,7 @@ func CleanSession(metaPath string, etcdEndpoints []string, sessionSuffix []strin
 	for _, key := range keys {
 		_, _ = etcdCli.Delete(ctx, key)
 	}
-	log.Info("clean sessions from etcd", zap.Any("keys", keys))
+	log.Ctx(ctx).Info("clean sessions from etcd", zap.Any("keys", keys))
 	return nil
 }
 
@@ -259,6 +259,7 @@ func getSessionPaths(ctx context.Context, client *clientv3.Client, metaPath stri
 
 // filterUnmatchedKey skip active keys that don't match completed key, the latest active key may from standby server
 func addActiveKeySuffix(ctx context.Context, client *clientv3.Client, sessionPathPrefix string, sessionSuffix []string) []string {
+	log := log.Ctx(ctx)
 	suffixSet := lo.SliceToMap(sessionSuffix, func(t string) (string, struct{}) {
 		return t, struct{}{}
 	})

cmd/tools/config/generate.go

@@ -1,6 +1,7 @@
 package main
 
 import (
+	"context"
 	"encoding/csv"
 	"fmt"
 	"io"
@@ -59,7 +60,7 @@ func collectRecursive(params *paramtable.ComponentParam, data *[]DocContent, val
 	if val.Kind() != reflect.Struct {
 		return
 	}
-	log.Debug("enter", zap.Any("variable", val.String()))
+	log.Ctx(context.TODO()).Debug("enter", zap.Any("variable", val.String()))
 	for j := 0; j < val.NumField(); j++ {
 		subVal := val.Field(j)
 		tag := val.Type().Field(j).Tag
@@ -71,11 +72,11 @@ func collectRecursive(params *paramtable.ComponentParam, data *[]DocContent, val
 			if strings.HasPrefix(item.DefaultValue, "\"") && strings.HasSuffix(item.DefaultValue, "\"") {
 				defaultValue = fmt.Sprintf("\"%s\"", defaultValue)
 			}
-			log.Debug("got key", zap.String("key", item.Key), zap.Any("value", defaultValue), zap.String("variable", val.Type().Field(j).Name))
+			log.Ctx(context.TODO()).Debug("got key", zap.String("key", item.Key), zap.Any("value", defaultValue), zap.String("variable", val.Type().Field(j).Name))
 			*data = append(*data, DocContent{item.Key, defaultValue, item.Version, refreshable, item.Export, item.Doc})
 		} else if t == "paramtable.ParamGroup" {
 			item := subVal.Interface().(paramtable.ParamGroup)
-			log.Debug("got key", zap.String("key", item.KeyPrefix), zap.String("variable", val.Type().Field(j).Name))
+			log.Ctx(context.TODO()).Debug("got key", zap.String("key", item.KeyPrefix), zap.String("variable", val.Type().Field(j).Name))
 			refreshable := tag.Get("refreshable")
 
 			// Sort group items to stablize the output order
@@ -87,7 +88,7 @@ func collectRecursive(params *paramtable.ComponentParam, data *[]DocContent, val
 			sort.Strings(keys)
 			for _, key := range keys {
 				value := m[key]
-				log.Debug("got group entry", zap.String("key", key), zap.String("value", value))
+				log.Ctx(context.TODO()).Debug("got group entry", zap.String("key", key), zap.String("value", value))
 				*data = append(*data, DocContent{fmt.Sprintf("%s%s", item.KeyPrefix, key), quoteIfNeeded(value), item.Version, refreshable, item.Export, item.GetDoc(key)})
 			}
 		} else {
@@ -148,7 +149,7 @@ func (m *YamlMarshaller) writeYamlRecursive(data []DocContent, level int) {
 	for _, key := range keys {
 		contents, ok := topLevels.Get(key)
 		if !ok {
-			log.Debug("didnot found config for " + key)
+			log.Ctx(context.TODO()).Debug("didnot found config for " + key)
 			continue
 		}
 		content := contents[0]

cmd/tools/migration/mmap/tool/main.go

@@ -136,7 +136,7 @@ func prepareRootCoordMeta(ctx context.Context, allocator tso.Allocator) rootcoor
 		}
 		catalog = &kvmetestore.Catalog{Txn: metaKV, Snapshot: ss}
 	case util.MetaStoreTypeTiKV:
-		log.Info("Using tikv as meta storage.")
+		log.Ctx(ctx).Info("Using tikv as meta storage.")
 		var metaKV kv.MetaKv
 		var ss *kvmetestore.SuffixSnapshot
 		var err error

configs/milvus.yaml

@@ -873,6 +873,7 @@ common:
   bloomFilterType: BlockedBloomFilter # bloom filter type, support BasicBloomFilter and BlockedBloomFilter
   maxBloomFalsePositive: 0.001 # max false positive rate for bloom filter
   bloomFilterApplyBatchSize: 1000 # batch size when to apply pk to bloom filter
+  collectionReplicateEnable: false # Whether to enable collection replication.
   usePartitionKeyAsClusteringKey: false # if true, do clustering compaction and segment prune on partition key field
   useVectorAsClusteringKey: false # if true, do clustering compaction and segment prune on vector field
   enableVectorClusteringKey: false # if true, enable vector clustering key and vector clustering compaction

internal/coordinator/coordclient/registry.go

@@ -66,7 +66,7 @@ func RegisterQueryCoordServer(server querypb.QueryCoordServer) {
 	}
 	newLocalClient := grpcclient.NewLocalGRPCClient(&querypb.QueryCoord_ServiceDesc, server, querypb.NewQueryCoordClient)
 	glocalClient.queryCoordClient.Set(&nopCloseQueryCoordClient{newLocalClient})
-	log.Info("register query coord server", zap.Any("enableLocalClient", enableLocal))
+	log.Ctx(context.TODO()).Info("register query coord server", zap.Any("enableLocalClient", enableLocal))
 }
 
 // RegsterDataCoordServer register data coord server
@@ -76,7 +76,7 @@ func RegisterDataCoordServer(server datapb.DataCoordServer) {
 	}
 	newLocalClient := grpcclient.NewLocalGRPCClient(&datapb.DataCoord_ServiceDesc, server, datapb.NewDataCoordClient)
 	glocalClient.dataCoordClient.Set(&nopCloseDataCoordClient{newLocalClient})
-	log.Info("register data coord server", zap.Any("enableLocalClient", enableLocal))
+	log.Ctx(context.TODO()).Info("register data coord server", zap.Any("enableLocalClient", enableLocal))
 }
 
 // RegisterRootCoordServer register root coord server
@@ -86,7 +86,7 @@ func RegisterRootCoordServer(server rootcoordpb.RootCoordServer) {
 	}
 	newLocalClient := grpcclient.NewLocalGRPCClient(&rootcoordpb.RootCoord_ServiceDesc, server, rootcoordpb.NewRootCoordClient)
 	glocalClient.rootCoordClient.Set(&nopCloseRootCoordClient{newLocalClient})
-	log.Info("register root coord server", zap.Any("enableLocalClient", enableLocal))
+	log.Ctx(context.TODO()).Info("register root coord server", zap.Any("enableLocalClient", enableLocal))
 }
 
 // GetQueryCoordClient return query coord client

internal/core/CMakeLists.txt

@@ -148,7 +148,9 @@ if ( APPLE )
             "-DELPP_THREAD_SAFE"
             "-fopenmp"
             "-pedantic"
-            "-Wno-all"
+            "-Wall"
+            "-Wno-gnu-zero-variadic-macro-arguments"
+            "-Wno-variadic-macros"
             "-DBOOST_STACKTRACE_GNU_SOURCE_NOT_REQUIRED=1"
             )
 endif ()

internal/core/src/common/Chunk.h

@@ -243,7 +243,7 @@ class SparseFloatVectorChunk : public Chunk {
         for (int i = 0; i < row_nums; i++) {
             vec_[i] = {(offsets_ptr[i + 1] - offsets_ptr[i]) /
                            knowhere::sparse::SparseRow<float>::element_size(),
-                       (uint8_t*)(data + offsets_ptr[i]),
+                       reinterpret_cast<uint8_t*>(data + offsets_ptr[i]),
                        false};
             dim_ = std::max(dim_, vec_[i].dim());
         }