update flush response

wayblink · Dec 1, 2023 · 5ed2244 · 5ed2244
1 parent af8aa51
commit 5ed2244
Show file tree

Hide file tree

Showing 2 changed files with 10 additions and 4 deletions.
diff --git a/client/client.go b/client/client.go
@@ -14,6 +14,7 @@ package client
 
 import (
 	"context"
+	"github.com/milvus-io/milvus-proto/go-api/v2/milvuspb"
 	"time"
 
 	"google.golang.org/grpc"
@@ -127,7 +128,7 @@ type Client interface {
 	Flush(ctx context.Context, collName string, async bool) error
 	// FlushV2 flush collection, specified, return newly sealed segmentIds, all flushed segmentIds of the collection, seal time and error
 	// currently it is only used in milvus-backup(https://github.com/zilliztech/milvus-backup)
-	FlushV2(ctx context.Context, collName string, async bool) ([]int64, []int64, int64, map[string]string, error)
+	FlushV2(ctx context.Context, collName string, async bool) ([]int64, []int64, int64, map[string]milvuspb.MsgPosition, error)
 	// DeleteByPks deletes entries related to provided primary keys
 	DeleteByPks(ctx context.Context, collName string, partitionName string, ids entity.Column) error
 	// Delete deletes entries match expression

diff --git a/client/insert.go b/client/insert.go
@@ -198,7 +198,7 @@ func (c *GrpcClient) Flush(ctx context.Context, collName string, async bool) err
 
 // Flush force collection to flush memory records into storage
 // in sync mode, flush will wait all segments to be flushed
-func (c *GrpcClient) FlushV2(ctx context.Context, collName string, async bool) ([]int64, []int64, int64, map[string]string, error) {
+func (c *GrpcClient) FlushV2(ctx context.Context, collName string, async bool) ([]int64, []int64, int64, map[string]milvuspb.MsgPosition, error) {
 	if c.Service == nil {
 		return nil, nil, 0, nil, ErrClientNotReady
 	}
@@ -246,9 +246,14 @@ func (c *GrpcClient) FlushV2(ctx context.Context, collName string, async bool) (
 			}
 		}
 	}
-	channelCPEntities := make(map[string]string, len(channelCPs))
+	channelCPEntities := make(map[string]milvuspb.MsgPosition, len(channelCPs))
 	for k, v := range channelCPs {
-		channelCPEntities[k] = Base64MsgPosition(v)
+		channelCPEntities[k] = milvuspb.MsgPosition{
+			ChannelName: v.GetChannelName(),
+			MsgID:       v.GetMsgID(),
+			MsgGroup:    v.GetMsgGroup(),
+			Timestamp:   v.GetTimestamp(),
+		}
 	}
 	return resp.GetCollSegIDs()[collName].GetData(), resp.GetFlushCollSegIDs()[collName].GetData(), resp.GetCollSealTimes()[collName], channelCPEntities, nil
 }