Repository: milvus-io/milvus-sdk-go
Pull request #810 (2024-08-21 05:04:11)

fix: Make `MetaCache` field of GrpcClient
Related to #809

sre-ci-robot: [APPROVALNOTIFIER] This PR is **APPROVED**

This pull-request has been approved by: *<a href="https://github.com/milvus-io/milvus-sdk-go/pull/810#" title="Author self-approved">congqixia</a>*

The full list of commands accepted by this bot can be found [here](https://go.k8s.io/bot-commands?repo=milvus-io%2Fmilvus-sdk-go).

The pull request process is described [here](https://git.k8s.io/community/contributors/guide/owners.md#the-code-review-process)

<details >
Needs approval from an approver in each of these files:

- ~~[OWNERS](https://github.com/milvus-io/milvus-sdk-go/blob/master/OWNERS)~~ [congqixia]

Approvers can indicate their approval by writing `/approve` in a comment
Approvers can cancel approval by writing `/approve cancel` in a comment
</details>
<!-- META={"approvers":[]} -->

Patch:
diff --git a/client/collection.go b/client/collection.go
index cbf6b58..359e250 100644
--- a/client/collection.go
+++ b/client/collection.go
@@ -292,7 +292,7 @@ func (c *GrpcClient) DescribeCollection(ctx context.Context, collName string) (*
 		Schema:           collection.Schema,
 		ConsistencyLevel: collection.ConsistencyLevel,
 	}
-	MetaCache.setCollectionInfo(collName, &colInfo)
+	c.cache.setCollectionInfo(collName, &colInfo)
 	return collection, nil
 }
 
@@ -317,7 +317,7 @@ func (c *GrpcClient) DropCollection(ctx context.Context, collName string, opts .
 	}
 	err = handleRespStatus(resp)
 	if err == nil {
-		MetaCache.setCollectionInfo(collName, nil)
+		c.cache.setCollectionInfo(collName, nil)
 	}
 	return err
 }
diff --git a/client/data.go b/client/data.go
index 68a7357..0e61a71 100644
--- a/client/data.go
+++ b/client/data.go
@@ -43,18 +43,11 @@ func (c *GrpcClient) HybridSearch(ctx context.Context, collName string, partitio
 		return nil, ErrClientNotReady
 	}
 
-	var schema *entity.Schema
-	collInfo, ok := MetaCache.getCollectionInfo(collName)
-	if !ok {
-		coll, err := c.DescribeCollection(ctx, collName)
-		if err != nil {
-			return nil, err
-		}
-		schema = coll.Schema
-		collInfo, _ = MetaCache.getCollectionInfo(collName)
-	} else {
-		schema = collInfo.Schema
+	collInfo, err := c.getCollectionInfo(ctx, collName)
+	if err != nil {
+		return nil, err
 	}
+	schema := collInfo.Schema
 
 	sReqs := make([]*milvuspb.SearchRequest, 0, len(subRequests))
 	nq := 0
@@ -103,19 +96,13 @@ func (c *GrpcClient) Search(ctx context.Context, collName string, partitions []s
 	if c.Service == nil {
 		return []SearchResult{}, ErrClientNotReady
 	}
-	var schema *entity.Schema
-	collInfo, ok := MetaCache.getCollectionInfo(collName)
-	if !ok {
-		coll, err := c.DescribeCollection(ctx, collName)
-		if err != nil {
-			return nil, err
-		}
-		schema = coll.Schema
-	} else {
-		schema = collInfo.Schema
+	collInfo, err := c.getCollectionInfo(ctx, collName)
+	if err != nil {
+		return nil, err
 	}
+	schema := collInfo.Schema
 
-	option, err := makeSearchQueryOption(collName, opts...)
+	option, err := c.makeSearchQueryOption(collName, opts...)
 	if err != nil {
 		return nil, err
 	}
@@ -320,19 +307,13 @@ func (c *GrpcClient) Query(ctx context.Context, collectionName string, partition
 		return nil, ErrClientNotReady
 	}
 
-	var sch *entity.Schema
-	collInfo, ok := MetaCache.getCollectionInfo(collectionName)
-	if !ok {
-		coll, err := c.DescribeCollection(ctx, collectionName)
-		if err != nil {
-			return nil, err
-		}
-		sch = coll.Schema
-	} else {
-		sch = collInfo.Schema
+	collInfo, err := c.getCollectionInfo(ctx, collectionName)
+	if err != nil {
+		return nil, err
 	}
+	sch := collInfo.Schema
 
-	option, err := makeSearchQueryOption(collectionName, opts...)
+	option, err := c.makeSearchQueryOption(collectionName, opts...)
 	if err != nil {
 		return nil, err
 	}
diff --git a/client/database.go b/client/database.go
index d05136f..cfa6929 100644
--- a/client/database.go
+++ b/client/database.go
@@ -36,6 +36,7 @@ func (c *GrpcClient) UsingDatabase(ctx context.Context, dbName string) error {
 	if err != nil {
 		return err
 	}
+	c.cache.reset()
 
 	return nil
 }
diff --git a/client/grpc_client.go b/client/grpc_client.go
index c21aa4c..4817ee6 100644
--- a/client/grpc_client.go
+++ b/client/grpc_client.go
@@ -7,6 +7,8 @@ import (
 	"strconv"
 	"time"
 
+	"golang.org/x/sync/singleflight"
+
 	"github.com/milvus-io/milvus-proto/go-api/v2/commonpb"
 	"github.com/milvus-io/milvus-proto/go-api/v2/milvuspb"
 	"github.com/milvus-io/milvus-sdk-go/v2/common"
@@ -24,6 +26,9 @@ type GrpcClient struct {
 	Conn    *grpc.ClientConn             // grpc connection instance
 	Service milvuspb.MilvusServiceClient // Service client stub
 
+	cache *metaCache
+	sf    singleflight.Group
+
 	config *Config // No thread safety
 }
 
@@ -32,6 +37,8 @@ func (c *GrpcClient) connect(ctx context.Context, addr string, opts ...grpc.Dial
 	if addr == "" {
 		return fmt.Errorf("address is empty")
 	}
+
+	c.cache = newMetaCache()
 	conn, err := grpc.DialContext(ctx, addr, opts...)
 	if err != nil {
 		return err
@@ -92,6 +99,32 @@ func (c *GrpcClient) connectInternal(ctx context.Context) error {
 	return nil
 }
 
+func (c *GrpcClient) getCollectionInfo(ctx context.Context, collectionName string) (*collInfo, error) {
+	// try cache first
+	info, ok := c.cache.getCollectionInfo(collectionName)
+	if ok {
+		return info, nil
+	}
+
+	// use singleflight to fetch collection info
+	v, err, _ := c.sf.Do(collectionName, func() (interface{}, error) {
+		coll, err := c.DescribeCollection(ctx, collectionName)
+		if err != nil {
+			return nil, err
+		}
+		return &collInfo{
+			ID:               coll.ID,
+			Name:             coll.Name,
+			Schema:           coll.Schema,
+			ConsistencyLevel: coll.ConsistencyLevel,
+		}, nil
+	})
+	if err != nil {
+		return nil, err
+	}
+	return v.(*collInfo), nil
+}
+
 // Close close the connection
 func (c *GrpcClient) Close() error {
 	if c.Conn != nil {
diff --git a/client/insert.go b/client/insert.go
index 99436a8..205b99a 100644
--- a/client/insert.go
+++ b/client/insert.go
@@ -35,17 +35,11 @@ func (c *GrpcClient) Insert(ctx context.Context, collName string, partitionName
 	if c.Service == nil {
 		return nil, ErrClientNotReady
 	}
-	var schema *entity.Schema
-	collInfo, ok := MetaCache.getCollectionInfo(collName)
-	if !ok {
-		coll, err := c.DescribeCollection(ctx, collName)
-		if err != nil {
-			return nil, err
-		}
-		schema = coll.Schema
-	} else {
-		schema = collInfo.Schema
+	collInfo, err := c.getCollectionInfo(ctx, collName)
+	if err != nil {
+		return nil, err
 	}
+	schema := collInfo.Schema
 
 	// convert columns to field data
 	fieldsData, rowSize, err := c.processInsertColumns(schema, columns...)
@@ -70,7 +64,7 @@ func (c *GrpcClient) Insert(ctx context.Context, collName string, partitionName
 	if err := handleRespStatus(resp.GetStatus()); err != nil {
 		return nil, err
 	}
-	MetaCache.setSessionTs(collName, resp.Timestamp)
+	c.cache.setSessionTs(collName, resp.Timestamp)
 	// 3. parse id column
 	return entity.IDColumns(schema, resp.GetIDs(), 0, -1)
 }
@@ -332,7 +326,7 @@ func (c *GrpcClient) DeleteByPks(ctx context.Context, collName string, partition
 	if err != nil {
 		return err
 	}
-	MetaCache.setSessionTs(collName, resp.Timestamp)
+	c.cache.setSessionTs(collName, resp.Timestamp)
 	return nil
 }
 
@@ -370,7 +364,7 @@ func (c *GrpcClient) Delete(ctx context.Context, collName string, partitionName
 	if err != nil {
 		return err
 	}
-	MetaCache.setSessionTs(collName, resp.Timestamp)
+	c.cache.setSessionTs(collName, resp.Timestamp)
 	return nil
 }
 
@@ -382,17 +376,11 @@ func (c *GrpcClient) Upsert(ctx context.Context, collName string, partitionName
 	if c.Service == nil {
 		return nil, ErrClientNotReady
 	}
-	var schema *entity.Schema
-	collInfo, ok := MetaCache.getCollectionInfo(collName)
-	if !ok {
-		coll, err := c.DescribeCollection(ctx, collName)
-		if err != nil {
-			return nil, err
-		}
-		schema = coll.Schema
-	} else {
-		schema = collInfo.Schema
+	collInfo, err := c.getCollectionInfo(ctx, collName)
+	if err != nil {
+		return nil, err
 	}
+	schema := collInfo.Schema
 
 	fieldsData, rowSize, err := c.processInsertColumns(schema, columns...)
 	if err != nil {
@@ -416,7 +404,7 @@ func (c *GrpcClient) Upsert(ctx context.Context, collName string, partitionName
 	if err := handleRespStatus(resp.GetStatus()); err != nil {
 		return nil, err
 	}
-	MetaCache.setSessionTs(collName, resp.Timestamp)
+	c.cache.setSessionTs(collName, resp.Timestamp)
 	// 3. parse id column
 	return entity.IDColumns(schema, resp.GetIDs(), 0, -1)
 }
diff --git a/client/iterator.go b/client/iterator.go
index 7d47e4f..5cb92ff 100644
--- a/client/iterator.go
+++ b/client/iterator.go
@@ -12,17 +12,11 @@ import (
 
 func (c *GrpcClient) SearchIterator(ctx context.Context, opt *SearchIteratorOption) (*SearchIterator, error) {
 	collectionName := opt.collectionName
-	var sch *entity.Schema
-	collInfo, ok := MetaCache.getCollectionInfo(collectionName)
-	if !ok {
-		coll, err := c.DescribeCollection(ctx, collectionName)
-		if err != nil {
-			return nil, err
-		}
-		sch = coll.Schema
-	} else {
-		sch = collInfo.Schema
+	collInfo, err := c.getCollectionInfo(ctx, collectionName)
+	if err != nil {
+		return nil, err
 	}
+	sch := collInfo.Schema
 
 	var vectorField *entity.Field
 	for _, field := range sch.Fields {
@@ -52,7 +46,7 @@ func (c *GrpcClient) SearchIterator(ctx context.Context, opt *SearchIteratorOpti
 		expr:      opt.expr,
 	}
 
-	err := itr.init(ctx)
+	err = itr.init(ctx)
 	if err != nil {
 		return nil, err
 	}
@@ -202,17 +196,11 @@ func (itr *SearchIterator) Next(ctx context.Context) (*SearchResult, error) {
 
 func (c *GrpcClient) QueryIterator(ctx context.Context, opt *QueryIteratorOption) (*QueryIterator, error) {
 	collectionName := opt.collectionName
-	var sch *entity.Schema
-	collInfo, ok := MetaCache.getCollectionInfo(collectionName)
-	if !ok {
-		coll, err := c.DescribeCollection(ctx, collectionName)
-		if err != nil {
-			return nil, err
-		}
-		sch = coll.Schema
-	} else {
-		sch = collInfo.Schema
+	collInfo, err := c.getCollectionInfo(ctx, collectionName)
+	if err != nil {
+		return nil, err
 	}
+	sch := collInfo.Schema
 
 	itr := &QueryIterator{
 		client: c,
@@ -227,7 +215,7 @@ func (c *GrpcClient) QueryIterator(ctx context.Context, opt *QueryIteratorOption
 		expr:      opt.expr,
 	}
 
-	err := itr.init(ctx)
+	err = itr.init(ctx)
 	if err != nil {
 		return nil, err
 	}
diff --git a/client/meta_cache.go b/client/meta_cache.go
index 5a98457..366feff 100644
--- a/client/meta_cache.go
+++ b/client/meta_cache.go
@@ -20,9 +20,11 @@ type collInfo struct {
 	ConsistencyLevel entity.ConsistencyLevel
 }
 
-var MetaCache = metaCache{
-	sessionTsMap: make(map[string]uint64),
-	collInfoMap:  make(map[string]collInfo),
+func newMetaCache() *metaCache {
+	return &metaCache{
+		sessionTsMap: make(map[string]uint64),
+		collInfoMap:  make(map[string]collInfo),
+	}
 }
 
 // timestampMap collects the last-write-timestamp of every collection, which is required by session consistency level.
diff --git a/client/options.go b/client/options.go
index 3867524..047b52e 100644
--- a/client/options.go
+++ b/client/options.go
@@ -122,6 +122,18 @@ func WithResourceGroups(rgs []string) LoadCollectionOption {
 	}
 }
 
+func WithLoadFields(fields ...string) LoadCollectionOption {
+	return func(req *milvuspb.LoadCollectionRequest) {
+		req.LoadFields = fields
+	}
+}
+
+func WithSkipDynamicFields(skip bool) LoadCollectionOption {
+	return func(req *milvuspb.LoadCollectionRequest) {
+		req.SkipLoadDynamicField = skip
+	}
+}
+
 // SearchQueryOption is an option of search/query request
 type SearchQueryOption struct {
 	// Consistency Level
@@ -206,11 +218,11 @@ func WithTravelTimestamp(_ uint64) SearchQueryOptionFunc {
 	return func(option *SearchQueryOption) {}
 }
 
-func makeSearchQueryOption(collName string, opts ...SearchQueryOptionFunc) (*SearchQueryOption, error) {
+func (c *GrpcClient) makeSearchQueryOption(collName string, opts ...SearchQueryOptionFunc) (*SearchQueryOption, error) {
 	opt := &SearchQueryOption{
 		ConsistencyLevel: entity.ClBounded, // default
 	}
-	info, ok := MetaCache.getCollectionInfo(collName)
+	info, ok := c.cache.getCollectionInfo(collName)
 	if ok {
 		opt.ConsistencyLevel = info.ConsistencyLevel
 	}
@@ -226,7 +238,7 @@ func makeSearchQueryOption(collName string, opts ...SearchQueryOptionFunc) (*Sea
 	case entity.ClStrong:
 		opt.GuaranteeTimestamp = StrongTimestamp
 	case entity.ClSession:
-		ts, ok := MetaCache.getSessionTs(collName)
+		ts, ok := c.cache.getSessionTs(collName)
 		if !ok {
 			ts = EventuallyTimestamp
 		}
diff --git a/client/row.go b/client/row.go
index 64ac5c0..6e36ab8 100644
--- a/client/row.go
+++ b/client/row.go
@@ -116,7 +116,7 @@ func (c *GrpcClient) InsertRows(ctx context.Context, collName string, partitionN
 	if err := handleRespStatus(resp.GetStatus()); err != nil {
 		return nil, err
 	}
-	MetaCache.setSessionTs(collName, resp.Timestamp)
+	c.cache.setSessionTs(collName, resp.Timestamp)
 	// 3. parse id column
 	return entity.IDColumns(coll.Schema, resp.GetIDs(), 0, -1)
 }
diff --git a/go.mod b/go.mod
index 7108f99..b739441 100644
--- a/go.mod
+++ b/go.mod
@@ -11,6 +11,7 @@ require (
 	github.com/stretchr/testify v1.8.1
 	github.com/tidwall/gjson v1.14.4
 	github.com/x448/float16 v0.8.4
+	golang.org/x/sync v0.8.0
 	google.golang.org/grpc v1.48.0
 	google.golang.org/grpc/examples v0.0.0-20220617181431-3e7b97febc7f
 )
diff --git a/go.sum b/go.sum
index 9ca5360..575eb0f 100644
--- a/go.sum
+++ b/go.sum
@@ -307,6 +307,8 @@ golang.org/x/sync v0.0.0-20201020160332-67f06af15bc9/go.mod h1:RxMgew5VJxzue5/jJ
 golang.org/x/sync v0.0.0-20210220032951-036812b2e83c/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20220722155255-886fb9371eb4/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.1.0/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
+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-20180909124046-d0be0721c37e/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20181205085412-a5c9d58dba9a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=


Test patch:
diff --git a/client/client_mock_test.go b/client/client_mock_test.go
index 38b7525..91215fc 100644
--- a/client/client_mock_test.go
+++ b/client/client_mock_test.go
@@ -29,7 +29,7 @@ type MockSuiteBase struct {
 	svr  *grpc.Server
 	mock *mocks.MilvusServiceServer
 
-	client Client
+	client *GrpcClient
 }
 
 func (s *MockSuiteBase) SetupSuite() {
@@ -71,7 +71,9 @@ func (s *MockSuiteBase) SetupTest() {
 	s.Require().NoError(err)
 	s.setupConnect()
 
-	s.client = c
+	grpcClient, ok := c.(*GrpcClient)
+	s.Require().True(ok)
+	s.client = grpcClient
 }
 
 func (s *MockSuiteBase) TearDownTest() {
@@ -80,8 +82,9 @@ func (s *MockSuiteBase) TearDownTest() {
 }
 
 func (s *MockSuiteBase) resetMock() {
-	MetaCache.reset()
+	// MetaCache.reset()
 	if s.mock != nil {
+		s.client.cache.reset()
 		s.mock.Calls = nil
 		s.mock.ExpectedCalls = nil
 		s.setupConnect()
diff --git a/client/collection_test.go b/client/collection_test.go
index c3a30ce..91baba0 100644
--- a/client/collection_test.go
+++ b/client/collection_test.go
@@ -276,10 +276,8 @@ func (s *CollectionSuite) TestCreateCollection() {
 			WithField(entity.NewField().WithName("dynamic").WithDataType(entity.FieldTypeJSON).WithIsDynamic(true))
 		for _, tc := range cases {
 			s.Run(tc.tag, func() {
-				grpcClient, ok := c.(*GrpcClient)
-				s.Require().True(ok)
-				grpcClient.config.addFlags(tc.flag)
-				defer grpcClient.config.resetFlags(tc.flag)
+				c.config.addFlags(tc.flag)
+				defer c.config.resetFlags(tc.flag)
 
 				err := c.CreateCollection(ctx, sch, 1)
 				s.Error(err)
diff --git a/client/options_test.go b/client/options_test.go
index 23f1dc7..0d0d6bc 100644
--- a/client/options_test.go
+++ b/client/options_test.go
@@ -58,6 +58,9 @@ func TestLoadCollectionWithReplicaNumber(t *testing.T) {
 }
 
 func TestMakeSearchQueryOption(t *testing.T) {
+	cli := &GrpcClient{
+		cache: newMetaCache(),
+	}
 	c := &entity.Collection{
 		Name:             "999",
 		ConsistencyLevel: entity.ClStrong,
@@ -67,10 +70,10 @@ func TestMakeSearchQueryOption(t *testing.T) {
 		Name:             c.Name,
 		ConsistencyLevel: c.ConsistencyLevel,
 	}
-	MetaCache.setCollectionInfo(c.Name, &cInfo)
+	cli.cache.setCollectionInfo(c.Name, &cInfo)
 
 	t.Run("strong consistency", func(t *testing.T) {
-		opt, err := makeSearchQueryOption(c.Name)
+		opt, err := cli.makeSearchQueryOption(c.Name)
 		assert.Nil(t, err)
 		assert.NotNil(t, opt)
 		expected := &SearchQueryOption{
@@ -83,7 +86,7 @@ func TestMakeSearchQueryOption(t *testing.T) {
 	})
 
 	t.Run("ignore growing", func(t *testing.T) {
-		opt, err := makeSearchQueryOption(c.Name, WithIgnoreGrowing())
+		opt, err := cli.makeSearchQueryOption(c.Name, WithIgnoreGrowing())
 		assert.Nil(t, err)
 		assert.NotNil(t, opt)
 		expected := &SearchQueryOption{
@@ -96,7 +99,7 @@ func TestMakeSearchQueryOption(t *testing.T) {
 	})
 
 	t.Run("for tuning", func(t *testing.T) {
-		opt, err := makeSearchQueryOption(c.Name, WithForTuning())
+		opt, err := cli.makeSearchQueryOption(c.Name, WithForTuning())
 		assert.Nil(t, err)
 		assert.NotNil(t, opt)
 		expected := &SearchQueryOption{
@@ -109,7 +112,7 @@ func TestMakeSearchQueryOption(t *testing.T) {
 	})
 
 	t.Run("session consistency", func(t *testing.T) {
-		opt, err := makeSearchQueryOption(c.Name, WithSearchQueryConsistencyLevel(entity.ClSession))
+		opt, err := cli.makeSearchQueryOption(c.Name, WithSearchQueryConsistencyLevel(entity.ClSession))
 		assert.Nil(t, err)
 		assert.NotNil(t, opt)
 		expected := &SearchQueryOption{
@@ -120,8 +123,8 @@ func TestMakeSearchQueryOption(t *testing.T) {
 		}
 		assert.Equal(t, expected, opt)
 
-		MetaCache.setSessionTs(c.Name, 99)
-		opt, err = makeSearchQueryOption(c.Name, WithSearchQueryConsistencyLevel(entity.ClSession))
+		cli.cache.setSessionTs(c.Name, 99)
+		opt, err = cli.makeSearchQueryOption(c.Name, WithSearchQueryConsistencyLevel(entity.ClSession))
 		assert.Nil(t, err)
 		assert.NotNil(t, opt)
 		expected = &SearchQueryOption{
@@ -134,7 +137,7 @@ func TestMakeSearchQueryOption(t *testing.T) {
 	})
 
 	t.Run("bounded consistency", func(t *testing.T) {
-		opt, err := makeSearchQueryOption(c.Name, WithSearchQueryConsistencyLevel(entity.ClBounded))
+		opt, err := cli.makeSearchQueryOption(c.Name, WithSearchQueryConsistencyLevel(entity.ClBounded))
 		assert.Nil(t, err)
 		assert.NotNil(t, opt)
 		expected := &SearchQueryOption{
@@ -147,7 +150,7 @@ func TestMakeSearchQueryOption(t *testing.T) {
 	})
 
 	t.Run("eventually consistency", func(t *testing.T) {
-		opt, err := makeSearchQueryOption(c.Name, WithSearchQueryConsistencyLevel(entity.ClEventually))
+		opt, err := cli.makeSearchQueryOption(c.Name, WithSearchQueryConsistencyLevel(entity.ClEventually))
 		assert.Nil(t, err)
 		assert.NotNil(t, opt)
 		expected := &SearchQueryOption{
@@ -160,7 +163,7 @@ func TestMakeSearchQueryOption(t *testing.T) {
 	})
 
 	t.Run("customized consistency", func(t *testing.T) {
-		opt, err := makeSearchQueryOption(c.Name, WithSearchQueryConsistencyLevel(entity.ClCustomized), WithGuaranteeTimestamp(100))
+		opt, err := cli.makeSearchQueryOption(c.Name, WithSearchQueryConsistencyLevel(entity.ClCustomized), WithGuaranteeTimestamp(100))
 		assert.Nil(t, err)
 		assert.NotNil(t, opt)
 		expected := &SearchQueryOption{
@@ -173,7 +176,7 @@ func TestMakeSearchQueryOption(t *testing.T) {
 	})
 
 	t.Run("guarantee timestamp sanity check", func(t *testing.T) {
-		_, err := makeSearchQueryOption(c.Name, WithSearchQueryConsistencyLevel(entity.ClStrong), WithGuaranteeTimestamp(100))
+		_, err := cli.makeSearchQueryOption(c.Name, WithSearchQueryConsistencyLevel(entity.ClStrong), WithGuaranteeTimestamp(100))
 		assert.Error(t, err)
 	})
 }