query master service's with timestamp (#5275)

let meta support snapshot
so collection could query meta with timestamp

Resolves: #5219
Signed-off-by: Nyefu.chen <yefu.chen@zilliz.com>

internal/distributed/masterservice/masterservice_test.go

@@ -360,7 +360,7 @@ func TestGrpcService(t *testing.T) {
 	})
 
 	t.Run("describe collection", func(t *testing.T) {
-		collMeta, err := core.MetaTable.GetCollectionByName("testColl")
+		collMeta, err := core.MetaTable.GetCollectionByName("testColl", 0)
 		assert.Nil(t, err)
 		req := &milvuspb.DescribeCollectionRequest{
 			Base: &commonpb.MsgBase{
@@ -411,10 +411,10 @@ func TestGrpcService(t *testing.T) {
 		status, err := cli.CreatePartition(ctx, req)
 		assert.Nil(t, err)
 		assert.Equal(t, status.ErrorCode, commonpb.ErrorCode_Success)
-		collMeta, err := core.MetaTable.GetCollectionByName("testColl")
+		collMeta, err := core.MetaTable.GetCollectionByName("testColl", 0)
 		assert.Nil(t, err)
 		assert.Equal(t, len(collMeta.PartitionIDs), 2)
-		partMeta, err := core.MetaTable.GetPartitionByID(collMeta.PartitionIDs[1])
+		partMeta, err := core.MetaTable.GetPartitionByID(1, collMeta.PartitionIDs[1], 0)
 		assert.Nil(t, err)
 		assert.Equal(t, partMeta.PartitionName, "testPartition")
 
@@ -440,7 +440,7 @@ func TestGrpcService(t *testing.T) {
 	})
 
 	t.Run("show partition", func(t *testing.T) {
-		coll, err := core.MetaTable.GetCollectionByName("testColl")
+		coll, err := core.MetaTable.GetCollectionByName("testColl", 0)
 		assert.Nil(t, err)
 		req := &milvuspb.ShowPartitionsRequest{
 			Base: &commonpb.MsgBase{
@@ -461,10 +461,10 @@ func TestGrpcService(t *testing.T) {
 	})
 
 	t.Run("show segment", func(t *testing.T) {
-		coll, err := core.MetaTable.GetCollectionByName("testColl")
+		coll, err := core.MetaTable.GetCollectionByName("testColl", 0)
 		assert.Nil(t, err)
 		partID := coll.PartitionIDs[1]
-		part, err := core.MetaTable.GetPartitionByID(partID)
+		part, err := core.MetaTable.GetPartitionByID(1, partID, 0)
 		assert.Nil(t, err)
 		assert.Zero(t, len(part.SegmentIDs))
 		seg := &datapb.SegmentInfo{
@@ -474,7 +474,7 @@ func TestGrpcService(t *testing.T) {
 		}
 		core.DataServiceSegmentChan <- seg
 		time.Sleep(time.Millisecond * 100)
-		part, err = core.MetaTable.GetPartitionByID(partID)
+		part, err = core.MetaTable.GetPartitionByID(1, partID, 0)
 		assert.Nil(t, err)
 		assert.Equal(t, len(part.SegmentIDs), 1)
 
@@ -513,13 +513,13 @@ func TestGrpcService(t *testing.T) {
 				},
 			},
 		}
-		collMeta, err := core.MetaTable.GetCollectionByName("testColl")
+		collMeta, err := core.MetaTable.GetCollectionByName("testColl", 0)
 		assert.Nil(t, err)
 		assert.Equal(t, len(collMeta.FieldIndexes), 0)
 		rsp, err := cli.CreateIndex(ctx, req)
 		assert.Nil(t, err)
 		assert.Equal(t, rsp.ErrorCode, commonpb.ErrorCode_Success)
-		collMeta, err = core.MetaTable.GetCollectionByName("testColl")
+		collMeta, err = core.MetaTable.GetCollectionByName("testColl", 0)
 		assert.Nil(t, err)
 		assert.Equal(t, len(collMeta.FieldIndexes), 1)
 
@@ -535,7 +535,7 @@ func TestGrpcService(t *testing.T) {
 	})
 
 	t.Run("describe segment", func(t *testing.T) {
-		coll, err := core.MetaTable.GetCollectionByName("testColl")
+		coll, err := core.MetaTable.GetCollectionByName("testColl", 0)
 		assert.Nil(t, err)
 
 		req := &milvuspb.DescribeSegmentRequest{
@@ -575,10 +575,10 @@ func TestGrpcService(t *testing.T) {
 	})
 
 	t.Run("flush segment", func(t *testing.T) {
-		coll, err := core.MetaTable.GetCollectionByName("testColl")
+		coll, err := core.MetaTable.GetCollectionByName("testColl", 0)
 		assert.Nil(t, err)
 		partID := coll.PartitionIDs[1]
-		part, err := core.MetaTable.GetPartitionByID(partID)
+		part, err := core.MetaTable.GetPartitionByID(1, partID, 0)
 		assert.Nil(t, err)
 		assert.Equal(t, len(part.SegmentIDs), 1)
 		seg := &datapb.SegmentInfo{
@@ -588,7 +588,7 @@ func TestGrpcService(t *testing.T) {
 		}
 		core.DataServiceSegmentChan <- seg
 		time.Sleep(time.Millisecond * 100)
-		part, err = core.MetaTable.GetPartitionByID(partID)
+		part, err = core.MetaTable.GetPartitionByID(1, partID, 0)
 		assert.Nil(t, err)
 		assert.Equal(t, len(part.SegmentIDs), 2)
 		core.DataNodeSegmentFlushCompletedChan <- 1001
@@ -656,10 +656,10 @@ func TestGrpcService(t *testing.T) {
 		status, err := cli.DropPartition(ctx, req)
 		assert.Nil(t, err)
 		assert.Equal(t, status.ErrorCode, commonpb.ErrorCode_Success)
-		collMeta, err := core.MetaTable.GetCollectionByName("testColl")
+		collMeta, err := core.MetaTable.GetCollectionByName("testColl", 0)
 		assert.Nil(t, err)
 		assert.Equal(t, len(collMeta.PartitionIDs), 1)
-		partMeta, err := core.MetaTable.GetPartitionByID(collMeta.PartitionIDs[0])
+		partMeta, err := core.MetaTable.GetPartitionByID(1, collMeta.PartitionIDs[0], 0)
 		assert.Nil(t, err)
 		assert.Equal(t, partMeta.PartitionName, cms.Params.DefaultPartitionName)
 		assert.Equal(t, 2, len(collectionMetaCache))

internal/kv/kv.go

@@ -11,6 +11,10 @@
 
 package kv
 
+import (
+	"github.com/milvus-io/milvus/internal/util/typeutil"
+)
+
 type BaseKV interface {
 	Load(key string) (string, error)
 	MultiLoad(keys []string) ([]string, error)
@@ -30,3 +34,11 @@ type TxnKV interface {
 	MultiRemoveWithPrefix(keys []string) error
 	MultiSaveAndRemoveWithPrefix(saves map[string]string, removals []string) error
 }
+
+type SnapShotKV interface {
+	Save(key, value string) (typeutil.Timestamp, error)
+	Load(key string, ts typeutil.Timestamp) (string, error)
+	MultiSave(kvs map[string]string) (typeutil.Timestamp, error)
+	LoadWithPrefix(key string, ts typeutil.Timestamp) ([]string, []string, error)
+	MultiSaveAndRemoveWithPrefix(saves map[string]string, removals []string) (typeutil.Timestamp, error)
+}

internal/masterservice/master_service.go

@@ -90,7 +90,6 @@ type Core struct {
 	cancel  context.CancelFunc
 	etcdCli *clientv3.Client
 	kvBase  *etcdkv.EtcdKV
-	metaKV  *etcdkv.EtcdKV
 
 	//setMsgStreams, receive time tick from proxy service time tick channel
 	ProxyTimeTickChan chan typeutil.Timestamp
@@ -191,9 +190,6 @@ func (c *Core) checkInit() error {
 	if c.etcdCli == nil {
 		return fmt.Errorf("etcdCli is nil")
 	}
-	if c.metaKV == nil {
-		return fmt.Errorf("metaKV is nil")
-	}
 	if c.kvBase == nil {
 		return fmt.Errorf("kvBase is nil")
 	}
@@ -307,7 +303,7 @@ func (c *Core) startDataServiceSegmentLoop() {
 			}
 			if seg == nil {
 				log.Warn("segment from data service is nil")
-			} else if err := c.MetaTable.AddSegment(seg); err != nil {
+			} else if _, err := c.MetaTable.AddSegment(seg); err != nil {
 				//what if master add segment failed, but data service success?
 				log.Warn("add segment info meta table failed ", zap.String("error", err.Error()))
 			} else {
@@ -387,7 +383,7 @@ func (c *Core) tsLoop() {
 }
 
 func (c *Core) setDdMsgSendFlag(b bool) error {
-	flag, err := c.MetaTable.client.Load(DDMsgSendPrefix)
+	flag, err := c.MetaTable.client.Load(DDMsgSendPrefix, 0)
 	if err != nil {
 		return err
 	}
@@ -398,9 +394,11 @@ func (c *Core) setDdMsgSendFlag(b bool) error {
 	}
 
 	if b {
-		return c.MetaTable.client.Save(DDMsgSendPrefix, "true")
+		_, err = c.MetaTable.client.Save(DDMsgSendPrefix, "true")
+		return err
 	}
-	return c.MetaTable.client.Save(DDMsgSendPrefix, "false")
+	_, err = c.MetaTable.client.Save(DDMsgSendPrefix, "false")
+	return err
 }
 
 func (c *Core) setMsgStreams() error {
@@ -808,7 +806,7 @@ func (c *Core) BuildIndex(segID typeutil.UniqueID, field *schemapb.FieldSchema,
 		BuildID:     bldID,
 		EnableIndex: enableIdx,
 	}
-	err = c.MetaTable.AddIndex(&seg)
+	_, err = c.MetaTable.AddIndex(&seg)
 	return err
 }
 
@@ -819,8 +817,23 @@ func (c *Core) Init() error {
 			if c.etcdCli, initError = clientv3.New(clientv3.Config{Endpoints: []string{Params.EtcdAddress}, DialTimeout: 5 * time.Second}); initError != nil {
 				return initError
 			}
-			c.metaKV = etcdkv.NewEtcdKV(c.etcdCli, Params.MetaRootPath)
-			if c.MetaTable, initError = NewMetaTable(c.metaKV); initError != nil {
+			tsAlloc := func() typeutil.Timestamp {
+				for {
+					var ts typeutil.Timestamp
+					var err error
+					if ts, err = c.tsoAllocator(1); err == nil {
+						return ts
+					}
+					time.Sleep(100 * time.Millisecond)
+					log.Debug("alloc time stamp error", zap.Error(err))
+				}
+			}
+			var ms *metaSnapshot
+			ms, initError = newMetaSnapshot(c.etcdCli, Params.MetaRootPath, TimestampPrefix, 1024, tsAlloc)
+			if initError != nil {
+				return initError
+			}
+			if c.MetaTable, initError = NewMetaTable(ms); initError != nil {
 				return initError
 			}
 			c.kvBase = etcdkv.NewEtcdKV(c.etcdCli, Params.KvRootPath)
@@ -876,13 +889,13 @@ func (c *Core) Init() error {
 }
 
 func (c *Core) reSendDdMsg(ctx context.Context) error {
-	flag, err := c.MetaTable.client.Load(DDMsgSendPrefix)
+	flag, err := c.MetaTable.client.Load(DDMsgSendPrefix, 0)
 	if err != nil || flag == "true" {
 		log.Debug("No un-successful DdMsg")
 		return nil
 	}
 
-	ddOpStr, err := c.MetaTable.client.Load(DDOperationPrefix)
+	ddOpStr, err := c.MetaTable.client.Load(DDOperationPrefix, 0)
 	if err != nil {
 		log.Debug("DdOperation key does not exist")
 		return nil
@@ -1638,27 +1651,29 @@ func (c *Core) AllocID(ctx context.Context, in *masterpb.AllocIDRequest) (*maste
 }
 
 func (c *Core) registerService(nodeName string, ip string) (<-chan *clientv3.LeaseKeepAliveResponse, error) {
-	respID, err := c.metaKV.Grant(5)
+	respID, err := c.etcdCli.Grant(c.ctx, 5)
 	if err != nil {
 		fmt.Printf("grant error %s\n", err)
 		return nil, err
 	}
 	c.session.NodeName = nodeName
 	c.session.IP = ip
-	c.session.LeaseID = respID
+	c.session.LeaseID = respID.ID
 
 	sessionJSON, err := json.Marshal(c.session)
 	if err != nil {
 		return nil, err
 	}
+	ctx, cancel := context.WithTimeout(c.ctx, RequestTimeout)
+	defer cancel()
 
-	err = c.metaKV.SaveWithLease(fmt.Sprintf("/node/%s", nodeName), string(sessionJSON), respID)
+	_, err = c.etcdCli.Put(ctx, fmt.Sprintf("%s/node/%s", Params.MetaRootPath, nodeName), string(sessionJSON), clientv3.WithLease(respID.ID))
 	if err != nil {
 		fmt.Printf("put lease error %s\n", err)
 		return nil, err
 	}
 
-	ch, err := c.metaKV.KeepAlive(respID)
+	ch, err := c.etcdCli.KeepAlive(c.ctx, respID.ID)
 	if err != nil {
 		fmt.Printf("keep alive error %s\n", err)
 		return nil, err

internal/masterservice/master_service_test.go

@@ -358,7 +358,7 @@ func TestMasterService(t *testing.T) {
 
 		createMsg, ok := (msg.Msgs[0]).(*msgstream.CreateCollectionMsg)
 		assert.True(t, ok)
-		createMeta, err := core.MetaTable.GetCollectionByName(collName)
+		createMeta, err := core.MetaTable.GetCollectionByName(collName, 0)
 		assert.Nil(t, err)
 		assert.Equal(t, createMeta.ID, createMsg.CollectionID)
 		assert.Equal(t, 1, len(createMeta.PartitionIDs))
@@ -414,15 +414,15 @@ func TestMasterService(t *testing.T) {
 		assert.True(t, ok)
 		createMsg, ok = (msg.Msgs[0]).(*msgstream.CreateCollectionMsg)
 		assert.True(t, ok)
-		createMeta, err = core.MetaTable.GetCollectionByName("testColl-again")
+		createMeta, err = core.MetaTable.GetCollectionByName("testColl-again", 0)
 		assert.Nil(t, err)
 		assert.Equal(t, createMeta.ID, createMsg.CollectionID)
 
 		// check DD operation info
-		flag, err := core.MetaTable.client.Load(DDMsgSendPrefix)
+		flag, err := core.MetaTable.client.Load(DDMsgSendPrefix, 0)
 		assert.Nil(t, err)
 		assert.Equal(t, "true", flag)
-		ddOpStr, err := core.MetaTable.client.Load(DDOperationPrefix)
+		ddOpStr, err := core.MetaTable.client.Load(DDOperationPrefix, 0)
 		assert.Nil(t, err)
 		var ddOp DdOperation
 		err = json.Unmarshal([]byte(ddOpStr), &ddOp)
@@ -490,7 +490,7 @@ func TestMasterService(t *testing.T) {
 	})
 
 	t.Run("describe collection", func(t *testing.T) {
-		collMeta, err := core.MetaTable.GetCollectionByName(collName)
+		collMeta, err := core.MetaTable.GetCollectionByName(collName, 0)
 		assert.Nil(t, err)
 		req := &milvuspb.DescribeCollectionRequest{
 			Base: &commonpb.MsgBase{
@@ -544,10 +544,10 @@ func TestMasterService(t *testing.T) {
 		status, err := core.CreatePartition(ctx, req)
 		assert.Nil(t, err)
 		assert.Equal(t, commonpb.ErrorCode_Success, status.ErrorCode)
-		collMeta, err := core.MetaTable.GetCollectionByName(collName)
+		collMeta, err := core.MetaTable.GetCollectionByName(collName, 0)
 		assert.Nil(t, err)
 		assert.Equal(t, 2, len(collMeta.PartitionIDs))
-		partMeta, err := core.MetaTable.GetPartitionByID(collMeta.PartitionIDs[1])
+		partMeta, err := core.MetaTable.GetPartitionByID(1, collMeta.PartitionIDs[1], 0)
 		assert.Nil(t, err)
 		assert.Equal(t, partName, partMeta.PartitionName)
 
@@ -563,10 +563,10 @@ func TestMasterService(t *testing.T) {
 		assert.Equal(t, collName, pm.GetCollArray()[0])
 
 		// check DD operation info
-		flag, err := core.MetaTable.client.Load(DDMsgSendPrefix)
+		flag, err := core.MetaTable.client.Load(DDMsgSendPrefix, 0)
 		assert.Nil(t, err)
 		assert.Equal(t, "true", flag)
-		ddOpStr, err := core.MetaTable.client.Load(DDOperationPrefix)
+		ddOpStr, err := core.MetaTable.client.Load(DDOperationPrefix, 0)
 		assert.Nil(t, err)
 		var ddOp DdOperation
 		err = json.Unmarshal([]byte(ddOpStr), &ddOp)
@@ -599,7 +599,7 @@ func TestMasterService(t *testing.T) {
 	})
 
 	t.Run("show partition", func(t *testing.T) {
-		coll, err := core.MetaTable.GetCollectionByName(collName)
+		coll, err := core.MetaTable.GetCollectionByName(collName, 0)
 		assert.Nil(t, err)
 		req := &milvuspb.ShowPartitionsRequest{
 			Base: &commonpb.MsgBase{
@@ -620,10 +620,10 @@ func TestMasterService(t *testing.T) {
 	})
 
 	t.Run("show segment", func(t *testing.T) {
-		coll, err := core.MetaTable.GetCollectionByName(collName)
+		coll, err := core.MetaTable.GetCollectionByName(collName, 0)
 		assert.Nil(t, err)
 		partID := coll.PartitionIDs[1]
-		part, err := core.MetaTable.GetPartitionByID(partID)
+		part, err := core.MetaTable.GetPartitionByID(1, partID, 0)
 		assert.Nil(t, err)
 		assert.Zero(t, len(part.SegmentIDs))
 
@@ -656,7 +656,7 @@ func TestMasterService(t *testing.T) {
 		assert.Nil(t, err)
 		time.Sleep(time.Second)
 
-		part, err = core.MetaTable.GetPartitionByID(partID)
+		part, err = core.MetaTable.GetPartitionByID(1, partID, 0)
 		assert.Nil(t, err)
 		assert.Equal(t, 1, len(part.SegmentIDs))
 
@@ -695,7 +695,7 @@ func TestMasterService(t *testing.T) {
 				},
 			},
 		}
-		collMeta, err := core.MetaTable.GetCollectionByName(collName)
+		collMeta, err := core.MetaTable.GetCollectionByName(collName, 0)
 		assert.Nil(t, err)
 		assert.Equal(t, 0, len(collMeta.FieldIndexes))
 
@@ -706,7 +706,7 @@ func TestMasterService(t *testing.T) {
 		files := im.getFileArray()
 		assert.Equal(t, 3, len(files))
 		assert.ElementsMatch(t, files, []string{"file0-100", "file1-100", "file2-100"})
-		collMeta, err = core.MetaTable.GetCollectionByName(collName)
+		collMeta, err = core.MetaTable.GetCollectionByName(collName, 0)
 		assert.Nil(t, err)
 		assert.Equal(t, 1, len(collMeta.FieldIndexes))
 		idxMeta, err := core.MetaTable.GetIndexByID(collMeta.FieldIndexes[0].IndexID)
@@ -720,7 +720,7 @@ func TestMasterService(t *testing.T) {
 	})
 
 	t.Run("describe segment", func(t *testing.T) {
-		coll, err := core.MetaTable.GetCollectionByName(collName)
+		coll, err := core.MetaTable.GetCollectionByName(collName, 0)
 		assert.Nil(t, err)
 
 		req := &milvuspb.DescribeSegmentRequest{
@@ -780,10 +780,10 @@ func TestMasterService(t *testing.T) {
 	})
 
 	t.Run("flush segment", func(t *testing.T) {
-		coll, err := core.MetaTable.GetCollectionByName(collName)
+		coll, err := core.MetaTable.GetCollectionByName(collName, 0)
 		assert.Nil(t, err)
 		partID := coll.PartitionIDs[1]
-		part, err := core.MetaTable.GetPartitionByID(partID)
+		part, err := core.MetaTable.GetPartitionByID(1, partID, 0)
 		assert.Nil(t, err)
 		assert.Equal(t, 1, len(part.SegmentIDs))
 
@@ -816,7 +816,7 @@ func TestMasterService(t *testing.T) {
 		assert.Nil(t, err)
 		time.Sleep(time.Second)
 
-		part, err = core.MetaTable.GetPartitionByID(partID)
+		part, err = core.MetaTable.GetPartitionByID(1, partID, 0)
 		assert.Nil(t, err)
 		assert.Equal(t, 2, len(part.SegmentIDs))
 
@@ -875,7 +875,7 @@ func TestMasterService(t *testing.T) {
 			},
 		}
 
-		collMeta, err := core.MetaTable.GetCollectionByName(collName)
+		collMeta, err := core.MetaTable.GetCollectionByName(collName, 0)
 		assert.Nil(t, err)
 		assert.Equal(t, 1, len(collMeta.FieldIndexes))
 		oldIdx := collMeta.FieldIndexes[0].IndexID
@@ -885,7 +885,7 @@ func TestMasterService(t *testing.T) {
 		assert.Equal(t, commonpb.ErrorCode_Success, rsp.ErrorCode)
 		time.Sleep(time.Second)
 
-		collMeta, err = core.MetaTable.GetCollectionByName(collName)
+		collMeta, err = core.MetaTable.GetCollectionByName(collName, 0)
 		assert.Nil(t, err)
 		assert.Equal(t, 2, len(collMeta.FieldIndexes))
 		assert.Equal(t, oldIdx, collMeta.FieldIndexes[0].IndexID)
@@ -943,16 +943,16 @@ func TestMasterService(t *testing.T) {
 			CollectionName: collName,
 			PartitionName:  partName,
 		}
-		collMeta, err := core.MetaTable.GetCollectionByName(collName)
+		collMeta, err := core.MetaTable.GetCollectionByName(collName, 0)
 		assert.Nil(t, err)
 		dropPartID := collMeta.PartitionIDs[1]
 		status, err := core.DropPartition(ctx, req)
 		assert.Nil(t, err)
 		assert.Equal(t, commonpb.ErrorCode_Success, status.ErrorCode)
-		collMeta, err = core.MetaTable.GetCollectionByName(collName)
+		collMeta, err = core.MetaTable.GetCollectionByName(collName, 0)
 		assert.Nil(t, err)
 		assert.Equal(t, 1, len(collMeta.PartitionIDs))
-		partMeta, err := core.MetaTable.GetPartitionByID(collMeta.PartitionIDs[0])
+		partMeta, err := core.MetaTable.GetPartitionByID(1, collMeta.PartitionIDs[0], 0)
 		assert.Nil(t, err)
 		assert.Equal(t, Params.DefaultPartitionName, partMeta.PartitionName)
 
@@ -968,10 +968,10 @@ func TestMasterService(t *testing.T) {
 		assert.Equal(t, collName, pm.GetCollArray()[1])
 
 		// check DD operation info
-		flag, err := core.MetaTable.client.Load(DDMsgSendPrefix)
+		flag, err := core.MetaTable.client.Load(DDMsgSendPrefix, 0)
 		assert.Nil(t, err)
 		assert.Equal(t, "true", flag)
-		ddOpStr, err := core.MetaTable.client.Load(DDOperationPrefix)
+		ddOpStr, err := core.MetaTable.client.Load(DDOperationPrefix, 0)
 		assert.Nil(t, err)
 		var ddOp DdOperation
 		err = json.Unmarshal([]byte(ddOpStr), &ddOp)
@@ -996,7 +996,7 @@ func TestMasterService(t *testing.T) {
 			DbName:         dbName,
 			CollectionName: collName,
 		}
-		collMeta, err := core.MetaTable.GetCollectionByName(collName)
+		collMeta, err := core.MetaTable.GetCollectionByName(collName, 0)
 		assert.Nil(t, err)
 		status, err := core.DropCollection(ctx, req)
 		assert.Nil(t, err)
@@ -1042,10 +1042,10 @@ func TestMasterService(t *testing.T) {
 		assert.Equal(t, collName, collArray[2])
 
 		// check DD operation info
-		flag, err := core.MetaTable.client.Load(DDMsgSendPrefix)
+		flag, err := core.MetaTable.client.Load(DDMsgSendPrefix, 0)
 		assert.Nil(t, err)
 		assert.Equal(t, "true", flag)
-		ddOpStr, err := core.MetaTable.client.Load(DDOperationPrefix)
+		ddOpStr, err := core.MetaTable.client.Load(DDOperationPrefix, 0)
 		assert.Nil(t, err)
 		var ddOp DdOperation
 		err = json.Unmarshal([]byte(ddOpStr), &ddOp)
@@ -1657,10 +1657,6 @@ func TestCheckInit(t *testing.T) {
 	err = c.checkInit()
 	assert.NotNil(t, err)
 
-	c.metaKV = &etcdkv.EtcdKV{}
-	err = c.checkInit()
-	assert.NotNil(t, err)
-
 	c.kvBase = &etcdkv.EtcdKV{}
 	err = c.checkInit()
 	assert.NotNil(t, err)

internal/masterservice/meta_snapshot.go

+// Copyright (C) 2019-2020 Zilliz. All rights reserved.
+//
+// Licensed 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 masterservice
+
+import (
+	"context"
+	"fmt"
+	"path"
+	"strconv"
+	"sync"
+	"time"
+
+	"github.com/milvus-io/milvus/internal/log"
+	"github.com/milvus-io/milvus/internal/util/typeutil"
+	"go.etcd.io/etcd/clientv3"
+	"go.uber.org/zap"
+)
+
+const (
+	RequestTimeout = 10 * time.Second
+)
+
+type rtPair struct {
+	rev int64
+	ts  typeutil.Timestamp
+}
+
+type metaSnapshot struct {
+	cli          *clientv3.Client
+	root         string
+	tsKey        string
+	lock         sync.RWMutex
+	timeAllactor func() typeutil.Timestamp
+
+	ts2Rev []rtPair
+	minPos int
+	maxPos int
+	numTs  int
+}
+
+func newMetaSnapshot(cli *clientv3.Client, root, tsKey string, bufSize int, timeAllactor func() typeutil.Timestamp) (*metaSnapshot, error) {
+	if bufSize <= 0 {
+		bufSize = 1024
+	}
+	ms := &metaSnapshot{
+		cli:          cli,
+		root:         root,
+		tsKey:        tsKey,
+		lock:         sync.RWMutex{},
+		timeAllactor: timeAllactor,
+		ts2Rev:       make([]rtPair, bufSize),
+		minPos:       0,
+		maxPos:       0,
+		numTs:        0,
+	}
+	if err := ms.loadTs(); err != nil {
+		return nil, err
+	}
+	return ms, nil
+}
+
+func (ms *metaSnapshot) loadTs() error {
+	ctx, cancel := context.WithTimeout(context.Background(), RequestTimeout)
+	defer cancel()
+
+	key := path.Join(ms.root, ms.tsKey)
+	resp, err := ms.cli.Get(ctx, key)
+	if err != nil {
+		return err
+	}
+	if len(resp.Kvs) <= 0 {
+		return nil
+	}
+	version := resp.Kvs[0].Version
+	revision := resp.Kvs[0].ModRevision
+	strTs := string(resp.Kvs[0].Value)
+	ts, err := strconv.ParseUint(strTs, 10, 64)
+	if err != nil {
+		return err
+	}
+	log.Info("load last ts", zap.Int64("version", version), zap.Int64("revision", revision))
+
+	ms.initTs(revision, ts)
+	for version--; version > 0; version-- {
+		if ms.numTs == len(ms.ts2Rev) {
+			break
+		}
+		revision--
+		resp, err = ms.cli.Get(ctx, key, clientv3.WithRev(revision))
+		if err != nil {
+			return err
+		}
+		if len(resp.Kvs) <= 0 {
+			return nil
+		}
+
+		curVer := resp.Kvs[0].Version
+		curRev := resp.Kvs[0].ModRevision
+		if curVer > version {
+			return nil
+		}
+		strTs := string(resp.Kvs[0].Value)
+		curTs, err := strconv.ParseUint(strTs, 10, 64)
+		if err != nil {
+			return err
+		}
+		if curTs >= ts {
+			return fmt.Errorf("timestamp go back, curTs=%d,ts=%d", curTs, ts)
+		}
+		ms.initTs(curRev, curTs)
+		ts = curTs
+		revision = curRev
+	}
+
+	return nil
+}
+
+func (ms *metaSnapshot) maxTs() typeutil.Timestamp {
+	return ms.ts2Rev[ms.maxPos].ts
+}
+
+func (ms *metaSnapshot) minTs() typeutil.Timestamp {
+	return ms.ts2Rev[ms.minPos].ts
+}
+
+func (ms *metaSnapshot) initTs(rev int64, ts typeutil.Timestamp) {
+	log.Debug("init meta snapshot ts", zap.Int64("rev", rev), zap.Uint64("ts", ts))
+	if ms.numTs == 0 {
+		ms.maxPos = len(ms.ts2Rev) - 1
+		ms.minPos = len(ms.ts2Rev) - 1
+		ms.numTs = 1
+		ms.ts2Rev[ms.maxPos].rev = rev
+		ms.ts2Rev[ms.maxPos].ts = ts
+	} else if ms.numTs < len(ms.ts2Rev) {
+		ms.minPos--
+		ms.numTs++
+		ms.ts2Rev[ms.minPos].rev = rev
+		ms.ts2Rev[ms.minPos].ts = ts
+	}
+}
+
+func (ms *metaSnapshot) putTs(rev int64, ts typeutil.Timestamp) {
+	log.Debug("put meta snapshto ts", zap.Int64("rev", rev), zap.Uint64("ts", ts))
+	ms.maxPos++
+	if ms.maxPos == len(ms.ts2Rev) {
+		ms.maxPos = 0
+	}
+
+	ms.ts2Rev[ms.maxPos].rev = rev
+	ms.ts2Rev[ms.maxPos].ts = ts
+	if ms.numTs < len(ms.ts2Rev) {
+		ms.numTs++
+	} else {
+		ms.minPos++
+		if ms.minPos == len(ms.ts2Rev) {
+			ms.minPos = 0
+		}
+	}
+}
+
+func (ms *metaSnapshot) searchOnCache(ts typeutil.Timestamp, start, length int) int64 {
+	if length == 1 {
+		return ms.ts2Rev[start].rev
+	}
+	begin := start
+	end := begin + length
+	mid := (begin + end) / 2
+	for {
+		if ms.ts2Rev[mid].ts == ts {
+			return ms.ts2Rev[mid].rev
+		}
+		if mid == begin {
+			if ms.ts2Rev[mid].ts < ts || mid == start {
+				return ms.ts2Rev[mid].rev
+			}
+			return ms.ts2Rev[mid-1].rev
+		}
+		if ms.ts2Rev[mid].ts > ts {
+			end = mid
+		} else if ms.ts2Rev[mid].ts < ts {
+			begin = mid + 1
+		}
+		mid = (begin + end) / 2
+	}
+}
+
+func (ms *metaSnapshot) getRevOnCache(ts typeutil.Timestamp) int64 {
+	if ms.numTs == 0 {
+		return 0
+	}
+	if ts >= ms.ts2Rev[ms.maxPos].ts {
+		return ms.ts2Rev[ms.maxPos].rev
+	}
+	if ts < ms.ts2Rev[ms.minPos].ts {
+		return 0
+	}
+	if ms.maxPos > ms.minPos {
+		return ms.searchOnCache(ts, ms.minPos, ms.maxPos-ms.minPos+1)
+	}
+	topVal := ms.ts2Rev[len(ms.ts2Rev)-1]
+	botVal := ms.ts2Rev[0]
+	minVal := ms.ts2Rev[ms.minPos]
+	maxVal := ms.ts2Rev[ms.maxPos]
+	if ts >= topVal.ts && ts < botVal.ts {
+		return topVal.rev
+	} else if ts >= minVal.ts && ts < topVal.ts {
+		return ms.searchOnCache(ts, ms.minPos, len(ms.ts2Rev)-ms.minPos)
+	} else if ts >= botVal.ts && ts < maxVal.ts {
+		return ms.searchOnCache(ts, 0, ms.maxPos+1)
+	}
+
+	return 0
+}
+
+func (ms *metaSnapshot) getRevOnEtcd(ts typeutil.Timestamp, rev int64) int64 {
+	if rev < 2 {
+		return 0
+	}
+	ctx, cancel := context.WithTimeout(context.Background(), RequestTimeout)
+	defer cancel()
+
+	for rev--; rev >= 2; rev-- {
+		resp, err := ms.cli.Get(ctx, path.Join(ms.root, ms.tsKey), clientv3.WithRev(rev))
+		if err != nil {
+			log.Debug("get ts from etcd failed", zap.Error(err))
+			return 0
+		}
+		if len(resp.Kvs) <= 0 {
+			return 0
+		}
+		rev = resp.Kvs[0].ModRevision
+		curTs, err := strconv.ParseUint(string(resp.Kvs[0].Value), 10, 64)
+		if err != nil {
+			log.Debug("parse timestam error", zap.String("input", string(resp.Kvs[0].Value)), zap.Error(err))
+			return 0
+		}
+		if curTs <= ts {
+			return rev
+		}
+	}
+	return 0
+}
+
+func (ms *metaSnapshot) getRev(ts typeutil.Timestamp) (int64, error) {
+	rev := ms.getRevOnCache(ts)
+	if rev > 0 {
+		return rev, nil
+	}
+	rev = ms.ts2Rev[ms.minPos].rev
+	rev = ms.getRevOnEtcd(ts, rev)
+	if rev > 0 {
+		return rev, nil
+	}
+	return 0, fmt.Errorf("can't find revision on ts=%d", ts)
+}
+
+func (ms *metaSnapshot) Save(key, value string) (typeutil.Timestamp, error) {
+	ms.lock.Lock()
+	defer ms.lock.Unlock()
+	ctx, cancel := context.WithTimeout(context.Background(), RequestTimeout)
+	defer cancel()
+
+	ts := ms.timeAllactor()
+	strTs := strconv.FormatInt(int64(ts), 10)
+	resp, err := ms.cli.Txn(ctx).If().Then(
+		clientv3.OpPut(path.Join(ms.root, key), value),
+		clientv3.OpPut(path.Join(ms.root, ms.tsKey), strTs),
+	).Commit()
+	if err != nil {
+		return 0, err
+	}
+	ms.putTs(resp.Header.Revision, ts)
+
+	return ts, nil
+}
+
+func (ms *metaSnapshot) Load(key string, ts typeutil.Timestamp) (string, error) {
+	ms.lock.RLock()
+	defer ms.lock.RUnlock()
+	ctx, cancel := context.WithTimeout(context.Background(), RequestTimeout)
+	defer cancel()
+
+	var resp *clientv3.GetResponse
+	var err error
+	var rev int64
+	if ts == 0 {
+		resp, err = ms.cli.Get(ctx, path.Join(ms.root, key))
+		if err != nil {
+			return "", err
+		}
+	} else {
+		rev, err = ms.getRev(ts)
+		if err != nil {
+			return "", err
+		}
+		resp, err = ms.cli.Get(ctx, path.Join(ms.root, key), clientv3.WithRev(rev))
+		if err != nil {
+			return "", err
+		}
+	}
+	if len(resp.Kvs) == 0 {
+		return "", fmt.Errorf("there is no value on key = %s, ts = %d", key, ts)
+	}
+	return string(resp.Kvs[0].Value), nil
+}
+
+func (ms *metaSnapshot) MultiSave(kvs map[string]string) (typeutil.Timestamp, error) {
+	ms.lock.Lock()
+	defer ms.lock.Unlock()
+	ctx, cancel := context.WithTimeout(context.Background(), RequestTimeout)
+	defer cancel()
+
+	ts := ms.timeAllactor()
+	strTs := strconv.FormatInt(int64(ts), 10)
+	ops := make([]clientv3.Op, 0, len(kvs)+1)
+	for key, value := range kvs {
+		ops = append(ops, clientv3.OpPut(path.Join(ms.root, key), value))
+	}
+	ops = append(ops, clientv3.OpPut(path.Join(ms.root, ms.tsKey), strTs))
+	resp, err := ms.cli.Txn(ctx).If().Then(ops...).Commit()
+	if err != nil {
+		return 0, err
+	}
+	ms.putTs(resp.Header.Revision, ts)
+	return ts, nil
+}
+func (ms *metaSnapshot) LoadWithPrefix(key string, ts typeutil.Timestamp) ([]string, []string, error) {
+	ms.lock.RLock()
+	defer ms.lock.RUnlock()
+	ctx, cancel := context.WithTimeout(context.Background(), RequestTimeout)
+	defer cancel()
+
+	var resp *clientv3.GetResponse
+	var err error
+	var rev int64
+	if ts == 0 {
+		resp, err = ms.cli.Get(ctx, path.Join(ms.root, key), clientv3.WithPrefix(), clientv3.WithSort(clientv3.SortByKey, clientv3.SortAscend))
+		if err != nil {
+			return nil, nil, err
+		}
+	} else {
+		rev, err = ms.getRev(ts)
+		if err != nil {
+			return nil, nil, err
+		}
+		resp, err = ms.cli.Get(ctx, path.Join(ms.root, key), clientv3.WithPrefix(), clientv3.WithRev(rev), clientv3.WithSort(clientv3.SortByKey, clientv3.SortAscend))
+		if err != nil {
+			return nil, nil, err
+		}
+	}
+	keys := make([]string, 0, len(resp.Kvs))
+	values := make([]string, 0, len(resp.Kvs))
+	tk := path.Join(ms.root, "k")
+	prefixLen := len(tk) - 1
+	for _, kv := range resp.Kvs {
+		tk = string(kv.Key)
+		tk = tk[prefixLen:]
+		keys = append(keys, tk)
+		values = append(values, string(kv.Value))
+	}
+	return keys, values, nil
+}
+
+func (ms *metaSnapshot) MultiSaveAndRemoveWithPrefix(saves map[string]string, removals []string) (typeutil.Timestamp, error) {
+	ms.lock.Lock()
+	defer ms.lock.Unlock()
+	ctx, cancel := context.WithTimeout(context.Background(), RequestTimeout)
+	defer cancel()
+
+	ts := ms.timeAllactor()
+	strTs := strconv.FormatInt(int64(ts), 10)
+	ops := make([]clientv3.Op, 0, len(saves)+len(removals)+1)
+	for key, value := range saves {
+		ops = append(ops, clientv3.OpPut(path.Join(ms.root, key), value))
+	}
+	for _, key := range removals {
+		ops = append(ops, clientv3.OpDelete(path.Join(ms.root, key)))
+	}
+	ops = append(ops, clientv3.OpPut(path.Join(ms.root, ms.tsKey), strTs))
+	resp, err := ms.cli.Txn(ctx).If().Then(ops...).Commit()
+	if err != nil {
+		return 0, err
+	}
+	ms.putTs(resp.Header.Revision, ts)
+	return ts, nil
+}

internal/masterservice/meta_snapshot_test.go

+// Copyright (C) 2019-2020 Zilliz. All rights reserved.
+//
+// Licensed 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 masterservice
+
+import (
+	"context"
+	"fmt"
+	"math/rand"
+	"path"
+	"testing"
+	"time"
+
+	"github.com/milvus-io/milvus/internal/util/typeutil"
+	"github.com/stretchr/testify/assert"
+	"go.etcd.io/etcd/clientv3"
+)
+
+func TestMetaSnapshot(t *testing.T) {
+	rand.Seed(time.Now().UnixNano())
+	randVal := rand.Int()
+
+	Params.Init()
+	etcdAddr := Params.EtcdAddress
+	rootPath := fmt.Sprintf("/test/meta/%d", randVal)
+	tsKey := "timestamp"
+
+	etcdCli, err := clientv3.New(clientv3.Config{Endpoints: []string{etcdAddr}})
+	assert.Nil(t, err)
+	defer etcdCli.Close()
+
+	var vtso typeutil.Timestamp
+	ftso := func() typeutil.Timestamp {
+		return vtso
+	}
+
+	ms, err := newMetaSnapshot(etcdCli, rootPath, tsKey, 4, ftso)
+	assert.Nil(t, err)
+	assert.NotNil(t, ms)
+
+	for i := 0; i < 8; i++ {
+		vtso = typeutil.Timestamp(100 + i)
+		ts, err := ms.Save("abc", fmt.Sprintf("value-%d", i))
+		assert.Nil(t, err)
+		assert.Equal(t, vtso, ts)
+		_, err = etcdCli.Put(context.Background(), "other", fmt.Sprintf("other-%d", i))
+		assert.Nil(t, err)
+	}
+
+	ms, err = newMetaSnapshot(etcdCli, rootPath, tsKey, 4, ftso)
+	assert.Nil(t, err)
+	assert.NotNil(t, ms)
+}
+
+func TestSearchOnCache(t *testing.T) {
+	ms := &metaSnapshot{}
+	for i := 0; i < 8; i++ {
+		ms.ts2Rev = append(ms.ts2Rev,
+			rtPair{
+				rev: int64(i * 2),
+				ts:  typeutil.Timestamp(i * 2),
+			})
+	}
+	rev := ms.searchOnCache(9, 0, 8)
+	assert.Equal(t, int64(8), rev)
+	rev = ms.searchOnCache(1, 0, 2)
+	assert.Equal(t, int64(0), rev)
+	rev = ms.searchOnCache(1, 0, 8)
+	assert.Equal(t, int64(0), rev)
+	rev = ms.searchOnCache(14, 0, 8)
+	assert.Equal(t, int64(14), rev)
+	rev = ms.searchOnCache(0, 0, 8)
+	assert.Equal(t, int64(0), rev)
+}
+
+func TestGetRevOnCache(t *testing.T) {
+	ms := &metaSnapshot{}
+	ms.ts2Rev = make([]rtPair, 7)
+	ms.initTs(7, 16)
+	ms.initTs(6, 14)
+	ms.initTs(5, 12)
+	ms.initTs(4, 10)
+
+	var rev int64
+	rev = ms.getRevOnCache(17)
+	assert.Equal(t, int64(7), rev)
+	rev = ms.getRevOnCache(9)
+	assert.Equal(t, int64(0), rev)
+	rev = ms.getRevOnCache(10)
+	assert.Equal(t, int64(4), rev)
+	rev = ms.getRevOnCache(16)
+	assert.Equal(t, int64(7), rev)
+	rev = ms.getRevOnCache(15)
+	assert.Equal(t, int64(6), rev)
+	rev = ms.getRevOnCache(12)
+	assert.Equal(t, int64(5), rev)
+
+	ms.initTs(3, 8)
+	ms.initTs(2, 6)
+	assert.Equal(t, ms.maxPos, 6)
+	assert.Equal(t, ms.minPos, 1)
+
+	rev = ms.getRevOnCache(17)
+	assert.Equal(t, int64(7), rev)
+	rev = ms.getRevOnCache(9)
+	assert.Equal(t, int64(3), rev)
+	rev = ms.getRevOnCache(10)
+	assert.Equal(t, int64(4), rev)
+	rev = ms.getRevOnCache(16)
+	assert.Equal(t, int64(7), rev)
+	rev = ms.getRevOnCache(15)
+	assert.Equal(t, int64(6), rev)
+	rev = ms.getRevOnCache(12)
+	assert.Equal(t, int64(5), rev)
+	rev = ms.getRevOnCache(5)
+	assert.Equal(t, int64(0), rev)
+
+	ms.putTs(8, 18)
+	assert.Equal(t, ms.maxPos, 0)
+	assert.Equal(t, ms.minPos, 1)
+	for rev = 2; rev <= 7; rev++ {
+		ts := ms.getRevOnCache(typeutil.Timestamp(rev*2 + 3))
+		assert.Equal(t, rev, ts)
+	}
+	ms.putTs(9, 20)
+	assert.Equal(t, ms.maxPos, 1)
+	assert.Equal(t, ms.minPos, 2)
+	assert.Equal(t, ms.numTs, 7)
+
+	curMax := ms.maxPos
+	curMin := ms.minPos
+	for i := 10; i < 20; i++ {
+		ms.putTs(int64(i), typeutil.Timestamp(i*2+2))
+		curMax++
+		curMin++
+		if curMax == len(ms.ts2Rev) {
+			curMax = 0
+		}
+		if curMin == len(ms.ts2Rev) {
+			curMin = 0
+		}
+		assert.Equal(t, curMax, ms.maxPos)
+		assert.Equal(t, curMin, ms.minPos)
+	}
+
+	for i := 13; i < 20; i++ {
+		rev = ms.getRevOnCache(typeutil.Timestamp(i*2 + 2))
+		assert.Equal(t, int64(i), rev)
+		rev = ms.getRevOnCache(typeutil.Timestamp(i*2 + 3))
+		assert.Equal(t, int64(i), rev)
+	}
+	rev = ms.getRevOnCache(27)
+	assert.Zero(t, rev)
+}
+
+func TestGetRevOnEtcd(t *testing.T) {
+	ctx, cancel := context.WithTimeout(context.Background(), 1*time.Second)
+	defer cancel()
+	rand.Seed(time.Now().UnixNano())
+	randVal := rand.Int()
+
+	Params.Init()
+	etcdAddr := Params.EtcdAddress
+	rootPath := fmt.Sprintf("/test/meta/%d", randVal)
+	tsKey := "timestamp"
+	key := path.Join(rootPath, tsKey)
+
+	etcdCli, err := clientv3.New(clientv3.Config{Endpoints: []string{etcdAddr}})
+	assert.Nil(t, err)
+	defer etcdCli.Close()
+
+	ms := metaSnapshot{
+		cli:   etcdCli,
+		root:  rootPath,
+		tsKey: tsKey,
+	}
+	resp, err := etcdCli.Put(ctx, key, "100")
+	assert.Nil(t, err)
+	revList := []int64{}
+	tsList := []typeutil.Timestamp{}
+	revList = append(revList, resp.Header.Revision)
+	tsList = append(tsList, 100)
+	for i := 110; i < 200; i += 10 {
+		resp, err = etcdCli.Put(ctx, key, fmt.Sprintf("%d", i))
+		assert.Nil(t, err)
+		revList = append(revList, resp.Header.Revision)
+		tsList = append(tsList, typeutil.Timestamp(i))
+	}
+	lastRev := revList[len(revList)-1] + 1
+	for i, ts := range tsList {
+		rev := ms.getRevOnEtcd(ts, lastRev)
+		assert.Equal(t, revList[i], rev)
+	}
+	for i := 0; i < len(tsList); i++ {
+		rev := ms.getRevOnEtcd(tsList[i]+5, lastRev)
+		assert.Equal(t, revList[i], rev)
+	}
+	rev := ms.getRevOnEtcd(200, lastRev)
+	assert.Equal(t, lastRev-1, rev)
+	rev = ms.getRevOnEtcd(99, lastRev)
+	assert.Zero(t, rev)
+}
+
+func TestLoad(t *testing.T) {
+	rand.Seed(time.Now().UnixNano())
+	randVal := rand.Int()
+
+	Params.Init()
+	etcdAddr := Params.EtcdAddress
+	rootPath := fmt.Sprintf("/test/meta/%d", randVal)
+	tsKey := "timestamp"
+
+	etcdCli, err := clientv3.New(clientv3.Config{Endpoints: []string{etcdAddr}})
+	assert.Nil(t, err)
+	defer etcdCli.Close()
+
+	var vtso typeutil.Timestamp
+	ftso := func() typeutil.Timestamp {
+		return vtso
+	}
+
+	ms, err := newMetaSnapshot(etcdCli, rootPath, tsKey, 7, ftso)
+	assert.Nil(t, err)
+	assert.NotNil(t, ms)
+
+	for i := 0; i < 20; i++ {
+		vtso = typeutil.Timestamp(100 + i*5)
+		ts, err := ms.Save("key", fmt.Sprintf("value-%d", i))
+		assert.Nil(t, err)
+		assert.Equal(t, vtso, ts)
+	}
+	for i := 0; i < 20; i++ {
+		val, err := ms.Load("key", typeutil.Timestamp(100+i*5+2))
+		assert.Nil(t, err)
+		assert.Equal(t, val, fmt.Sprintf("value-%d", i))
+	}
+	val, err := ms.Load("key", 0)
+	assert.Nil(t, err)
+	assert.Equal(t, "value-19", val)
+
+	ms, err = newMetaSnapshot(etcdCli, rootPath, tsKey, 11, ftso)
+	assert.Nil(t, err)
+	assert.NotNil(t, ms)
+
+	for i := 0; i < 20; i++ {
+		val, err := ms.Load("key", typeutil.Timestamp(100+i*5+2))
+		assert.Nil(t, err)
+		assert.Equal(t, val, fmt.Sprintf("value-%d", i))
+	}
+}
+
+func TestMultiSave(t *testing.T) {
+	rand.Seed(time.Now().UnixNano())
+	randVal := rand.Int()
+
+	Params.Init()
+	etcdAddr := Params.EtcdAddress
+	rootPath := fmt.Sprintf("/test/meta/%d", randVal)
+	tsKey := "timestamp"
+
+	etcdCli, err := clientv3.New(clientv3.Config{Endpoints: []string{etcdAddr}})
+	assert.Nil(t, err)
+	defer etcdCli.Close()
+
+	var vtso typeutil.Timestamp
+	ftso := func() typeutil.Timestamp {
+		return vtso
+	}
+
+	ms, err := newMetaSnapshot(etcdCli, rootPath, tsKey, 7, ftso)
+	assert.Nil(t, err)
+	assert.NotNil(t, ms)
+
+	for i := 0; i < 20; i++ {
+		saves := map[string]string{"k1": fmt.Sprintf("v1-%d", i), "k2": fmt.Sprintf("v2-%d", i)}
+		vtso = typeutil.Timestamp(100 + i*5)
+		ts, err := ms.MultiSave(saves)
+		assert.Nil(t, err)
+		assert.Equal(t, vtso, ts)
+	}
+	for i := 0; i < 20; i++ {
+		keys, vals, err := ms.LoadWithPrefix("k", typeutil.Timestamp(100+i*5+2))
+		assert.Nil(t, err)
+		assert.Equal(t, len(keys), len(vals))
+		assert.Equal(t, len(keys), 2)
+		assert.Equal(t, keys[0], "k1")
+		assert.Equal(t, keys[1], "k2")
+		assert.Equal(t, vals[0], fmt.Sprintf("v1-%d", i))
+		assert.Equal(t, vals[1], fmt.Sprintf("v2-%d", i))
+	}
+	keys, vals, err := ms.LoadWithPrefix("k", 0)
+	assert.Nil(t, err)
+	assert.Equal(t, len(keys), len(vals))
+	assert.Equal(t, len(keys), 2)
+	assert.Equal(t, keys[0], "k1")
+	assert.Equal(t, keys[1], "k2")
+	assert.Equal(t, vals[0], "v1-19")
+	assert.Equal(t, vals[1], "v2-19")
+
+	ms, err = newMetaSnapshot(etcdCli, rootPath, tsKey, 11, ftso)
+	assert.Nil(t, err)
+	assert.NotNil(t, ms)
+
+	for i := 0; i < 20; i++ {
+		keys, vals, err := ms.LoadWithPrefix("k", typeutil.Timestamp(100+i*5+2))
+		assert.Nil(t, err)
+		assert.Equal(t, len(keys), len(vals))
+		assert.Equal(t, len(keys), 2)
+		assert.Equal(t, keys[0], "k1")
+		assert.Equal(t, keys[1], "k2")
+		assert.Equal(t, vals[0], fmt.Sprintf("v1-%d", i))
+		assert.Equal(t, vals[1], fmt.Sprintf("v2-%d", i))
+	}
+}
+
+func TestMultiSaveAndRemoveWithPrefix(t *testing.T) {
+	rand.Seed(time.Now().UnixNano())
+	randVal := rand.Int()
+
+	Params.Init()
+	etcdAddr := Params.EtcdAddress
+	rootPath := fmt.Sprintf("/test/meta/%d", randVal)
+	tsKey := "timestamp"
+
+	etcdCli, err := clientv3.New(clientv3.Config{Endpoints: []string{etcdAddr}})
+	assert.Nil(t, err)
+
+	var vtso typeutil.Timestamp
+	ftso := func() typeutil.Timestamp {
+		return vtso
+	}
+	defer etcdCli.Close()
+
+	ms, err := newMetaSnapshot(etcdCli, rootPath, tsKey, 7, ftso)
+	assert.Nil(t, err)
+	assert.NotNil(t, ms)
+
+	for i := 0; i < 20; i++ {
+		vtso = typeutil.Timestamp(100 + i*5)
+		ts, err := ms.Save(fmt.Sprintf("kd-%d", i), fmt.Sprintf("value-%d", i))
+		assert.Nil(t, err)
+		assert.Equal(t, vtso, ts)
+	}
+	for i := 20; i < 40; i++ {
+		sm := map[string]string{"ks": fmt.Sprintf("value-%d", i)}
+		dm := []string{fmt.Sprintf("kd-%d", i-20)}
+		vtso = typeutil.Timestamp(100 + i*5)
+		ts, err := ms.MultiSaveAndRemoveWithPrefix(sm, dm)
+		assert.Nil(t, err)
+		assert.Equal(t, vtso, ts)
+	}
+
+	for i := 0; i < 20; i++ {
+		val, err := ms.Load(fmt.Sprintf("kd-%d", i), typeutil.Timestamp(100+i*5+2))
+		assert.Nil(t, err)
+		assert.Equal(t, fmt.Sprintf("value-%d", i), val)
+		_, vals, err := ms.LoadWithPrefix("kd-", typeutil.Timestamp(100+i*5+2))
+		assert.Nil(t, err)
+		assert.Equal(t, i+1, len(vals))
+	}
+	for i := 20; i < 40; i++ {
+		val, err := ms.Load("ks", typeutil.Timestamp(100+i*5+2))
+		assert.Nil(t, err)
+		assert.Equal(t, fmt.Sprintf("value-%d", i), val)
+		_, vals, err := ms.LoadWithPrefix("kd-", typeutil.Timestamp(100+i*5+2))
+		assert.Nil(t, err)
+		assert.Equal(t, 39-i, len(vals))
+	}
+
+	ms, err = newMetaSnapshot(etcdCli, rootPath, tsKey, 11, ftso)
+	assert.Nil(t, err)
+	assert.NotNil(t, ms)
+
+	for i := 0; i < 20; i++ {
+		val, err := ms.Load(fmt.Sprintf("kd-%d", i), typeutil.Timestamp(100+i*5+2))
+		assert.Nil(t, err)
+		assert.Equal(t, fmt.Sprintf("value-%d", i), val)
+		_, vals, err := ms.LoadWithPrefix("kd-", typeutil.Timestamp(100+i*5+2))
+		assert.Nil(t, err)
+		assert.Equal(t, i+1, len(vals))
+	}
+	for i := 20; i < 40; i++ {
+		val, err := ms.Load("ks", typeutil.Timestamp(100+i*5+2))
+		assert.Nil(t, err)
+		assert.Equal(t, fmt.Sprintf("value-%d", i), val)
+		_, vals, err := ms.LoadWithPrefix("kd-", typeutil.Timestamp(100+i*5+2))
+		assert.Nil(t, err)
+		assert.Equal(t, 39-i, len(vals))
+	}
+}

internal/masterservice/task.go

@@ -222,7 +222,7 @@ func (t *CreateCollectionReqTask) Execute(ctx context.Context) error {
 		return err
 	}
 
-	err = t.core.MetaTable.AddCollection(&collInfo, &partInfo, idxInfo, ddOpStr)
+	_, err = t.core.MetaTable.AddCollection(&collInfo, &partInfo, idxInfo, ddOpStr)
 	if err != nil {
 		return err
 	}
@@ -266,7 +266,7 @@ func (t *DropCollectionReqTask) Execute(ctx context.Context) error {
 		return fmt.Errorf("drop collection, msg type = %s", commonpb.MsgType_name[int32(t.Type())])
 	}
 
-	collMeta, err := t.core.MetaTable.GetCollectionByName(t.Req.CollectionName)
+	collMeta, err := t.core.MetaTable.GetCollectionByName(t.Req.CollectionName, 0)
 	if err != nil {
 		return err
 	}
@@ -286,7 +286,7 @@ func (t *DropCollectionReqTask) Execute(ctx context.Context) error {
 		return err
 	}
 
-	err = t.core.MetaTable.DeleteCollection(collMeta.ID, ddOpStr)
+	_, err = t.core.MetaTable.DeleteCollection(collMeta.ID, ddOpStr)
 	if err != nil {
 		return err
 	}
@@ -336,7 +336,7 @@ func (t *HasCollectionReqTask) Execute(ctx context.Context) error {
 	if t.Type() != commonpb.MsgType_HasCollection {
 		return fmt.Errorf("has collection, msg type = %s", commonpb.MsgType_name[int32(t.Type())])
 	}
-	_, err := t.core.MetaTable.GetCollectionByName(t.Req.CollectionName)
+	_, err := t.core.MetaTable.GetCollectionByName(t.Req.CollectionName, 0)
 	if err == nil {
 		t.HasCollection = true
 	} else {
@@ -375,12 +375,12 @@ func (t *DescribeCollectionReqTask) Execute(ctx context.Context) error {
 	var err error
 
 	if t.Req.CollectionName != "" {
-		collInfo, err = t.core.MetaTable.GetCollectionByName(t.Req.CollectionName)
+		collInfo, err = t.core.MetaTable.GetCollectionByName(t.Req.CollectionName, 0)
 		if err != nil {
 			return err
 		}
 	} else {
-		collInfo, err = t.core.MetaTable.GetCollectionByID(t.Req.CollectionID)
+		collInfo, err = t.core.MetaTable.GetCollectionByID(t.Req.CollectionID, 0)
 		if err != nil {
 			return err
 		}
@@ -427,7 +427,7 @@ func (t *ShowCollectionReqTask) Execute(ctx context.Context) error {
 	if t.Type() != commonpb.MsgType_ShowCollections {
 		return fmt.Errorf("show collection, msg type = %s", commonpb.MsgType_name[int32(t.Type())])
 	}
-	coll, err := t.core.MetaTable.ListCollections()
+	coll, err := t.core.MetaTable.ListCollections(0)
 	if err != nil {
 		return err
 	}
@@ -460,7 +460,7 @@ func (t *CreatePartitionReqTask) Execute(ctx context.Context) error {
 	if t.Type() != commonpb.MsgType_CreatePartition {
 		return fmt.Errorf("create partition, msg type = %s", commonpb.MsgType_name[int32(t.Type())])
 	}
-	collMeta, err := t.core.MetaTable.GetCollectionByName(t.Req.CollectionName)
+	collMeta, err := t.core.MetaTable.GetCollectionByName(t.Req.CollectionName, 0)
 	if err != nil {
 		return err
 	}
@@ -486,7 +486,7 @@ func (t *CreatePartitionReqTask) Execute(ctx context.Context) error {
 		return err
 	}
 
-	err = t.core.MetaTable.AddPartition(collMeta.ID, t.Req.PartitionName, partID, ddOpStr)
+	_, err = t.core.MetaTable.AddPartition(collMeta.ID, t.Req.PartitionName, partID, ddOpStr)
 	if err != nil {
 		return err
 	}
@@ -528,11 +528,11 @@ func (t *DropPartitionReqTask) Execute(ctx context.Context) error {
 	if t.Type() != commonpb.MsgType_DropPartition {
 		return fmt.Errorf("drop partition, msg type = %s", commonpb.MsgType_name[int32(t.Type())])
 	}
-	collInfo, err := t.core.MetaTable.GetCollectionByName(t.Req.CollectionName)
+	collInfo, err := t.core.MetaTable.GetCollectionByName(t.Req.CollectionName, 0)
 	if err != nil {
 		return err
 	}
-	partInfo, err := t.core.MetaTable.GetPartitionByName(collInfo.ID, t.Req.PartitionName)
+	partInfo, err := t.core.MetaTable.GetPartitionByName(collInfo.ID, t.Req.PartitionName, 0)
 	if err != nil {
 		return err
 	}
@@ -554,7 +554,7 @@ func (t *DropPartitionReqTask) Execute(ctx context.Context) error {
 		return err
 	}
 
-	_, err = t.core.MetaTable.DeletePartition(collInfo.ID, t.Req.PartitionName, ddOpStr)
+	_, _, err = t.core.MetaTable.DeletePartition(collInfo.ID, t.Req.PartitionName, ddOpStr)
 	if err != nil {
 		return err
 	}
@@ -597,11 +597,11 @@ func (t *HasPartitionReqTask) Execute(ctx context.Context) error {
 	if t.Type() != commonpb.MsgType_HasPartition {
 		return fmt.Errorf("has partition, msg type = %s", commonpb.MsgType_name[int32(t.Type())])
 	}
-	coll, err := t.core.MetaTable.GetCollectionByName(t.Req.CollectionName)
+	coll, err := t.core.MetaTable.GetCollectionByName(t.Req.CollectionName, 0)
 	if err != nil {
 		return err
 	}
-	t.HasPartition = t.core.MetaTable.HasPartition(coll.ID, t.Req.PartitionName)
+	t.HasPartition = t.core.MetaTable.HasPartition(coll.ID, t.Req.PartitionName, 0)
 	return nil
 }
 
@@ -634,15 +634,15 @@ func (t *ShowPartitionReqTask) Execute(ctx context.Context) error {
 	var coll *etcdpb.CollectionInfo
 	var err error
 	if t.Req.CollectionName == "" {
-		coll, err = t.core.MetaTable.GetCollectionByID(t.Req.CollectionID)
+		coll, err = t.core.MetaTable.GetCollectionByID(t.Req.CollectionID, 0)
 	} else {
-		coll, err = t.core.MetaTable.GetCollectionByName(t.Req.CollectionName)
+		coll, err = t.core.MetaTable.GetCollectionByName(t.Req.CollectionName, 0)
 	}
 	if err != nil {
 		return err
 	}
 	for _, partID := range coll.PartitionIDs {
-		partMeta, err := t.core.MetaTable.GetPartitionByID(partID)
+		partMeta, err := t.core.MetaTable.GetPartitionByID(coll.ID, partID, 0)
 		if err != nil {
 			return err
 		}
@@ -678,7 +678,7 @@ func (t *DescribeSegmentReqTask) Execute(ctx context.Context) error {
 	if t.Type() != commonpb.MsgType_DescribeSegment {
 		return fmt.Errorf("describe segment, msg type = %s", commonpb.MsgType_name[int32(t.Type())])
 	}
-	coll, err := t.core.MetaTable.GetCollectionByID(t.Req.CollectionID)
+	coll, err := t.core.MetaTable.GetCollectionByID(t.Req.CollectionID, 0)
 	if err != nil {
 		return err
 	}
@@ -687,7 +687,7 @@ func (t *DescribeSegmentReqTask) Execute(ctx context.Context) error {
 		if exist {
 			break
 		}
-		partMeta, err := t.core.MetaTable.GetPartitionByID(partID)
+		partMeta, err := t.core.MetaTable.GetPartitionByID(coll.ID, partID, 0)
 		if err != nil {
 			return err
 		}
@@ -738,7 +738,7 @@ func (t *ShowSegmentReqTask) Execute(ctx context.Context) error {
 	if t.Type() != commonpb.MsgType_ShowSegments {
 		return fmt.Errorf("show segments, msg type = %s", commonpb.MsgType_name[int32(t.Type())])
 	}
-	coll, err := t.core.MetaTable.GetCollectionByID(t.Req.CollectionID)
+	coll, err := t.core.MetaTable.GetCollectionByID(t.Req.CollectionID, 0)
 	if err != nil {
 		return err
 	}
@@ -752,7 +752,7 @@ func (t *ShowSegmentReqTask) Execute(ctx context.Context) error {
 	if !exist {
 		return fmt.Errorf("partition id = %d not belong to collection id = %d", t.Req.PartitionID, t.Req.CollectionID)
 	}
-	partMeta, err := t.core.MetaTable.GetPartitionByID(t.Req.PartitionID)
+	partMeta, err := t.core.MetaTable.GetPartitionByID(coll.ID, t.Req.PartitionID, 0)
 	if err != nil {
 		return err
 	}
@@ -900,6 +900,6 @@ func (t *DropIndexReqTask) Execute(ctx context.Context) error {
 	if err != nil {
 		return err
 	}
-	_, _, err = t.core.MetaTable.DropIndex(t.Req.CollectionName, t.Req.FieldName, t.Req.IndexName)
+	_, _, _, err = t.core.MetaTable.DropIndex(t.Req.CollectionName, t.Req.FieldName, t.Req.IndexName)
 	return err
 }