使用query方法统计collection中entities数目报错

[gobigrassland]

使用sift特征，循环方式以固定batch插入并统计该collection中entities的数目，循环执行若干次能正常运行，执行到某次时就报grpc错误。不明白为什么调用collection.query会出现gprc 错误，开始若干次没有报错，之后就报错？

代码片段：

def create_collection_sift(collection_name):
    # fieldschema
    idx = FieldSchema(
        name="idx",
        dtype=DataType.INT64,
        is_primary=True)

    feature = FieldSchema(
        name="feature",
        dtype=DataType.FLOAT_VECTOR,
        dim=128)

    schema = CollectionSchema(
        fields=[idx,feature],
        description="Test sift search")

    collection = Collection(
        name=collection_name,
        schema=schema,
        shards_num=2,
        consistency_level="Strong")

def load_sift():
    x = np.memmap('data/bigann_base.bvecs', dtype='uint8', mode='r')
    d = x[:4].view('int32')[0]
    gallery = x.reshape(-1, d + 4)[:, 4:]
    x = np.memmap('data/bigann_query.bvecs', dtype='uint8', mode='r')
    d = x[:4].view('int32')[0]
    probe = x.reshape(-1, d + 4)[:, 4:]
    return probe, gallery

def count(collection_name):
    collection = Collection(collection_name)
    collection.load()
    res = collection.query(
        expr = "idx>=0",
        output_fields = ["idx"],
        consistency_level="Strong"
        )
    return res

def test_milvus_sift(cname, topk=5, num=100000):
    # load feature for testing
    probe_feat, gallery_feat = load_sift()
    probe_feat = preprocessing.normalize(probe_feat).astype(np.float32)
    gallery_feat = preprocessing.normalize(gallery_feat[:num,:]).astype(np.float32)
    # create a collection
    create_collection_sift(cname)
    collection = Collection(cname)
    t1 = time.time()
    batch_size = 100000
    num_batch = int(len(gallery_feat)/batch_size)
    for i in range(num_batch):
        data = [[ind for ind in range(i*batch_size, (i+1)*batch_size)], gallery_feat[i*batch_size:(i+1)*batch_size, :]]
        collection.insert(data)
        res = count(cname)
        print('Number of entities: ', len(res))
    batch_last = len(gallery_feat) - num_batch * batch_size
    if batch_last > 0:
        data = [[ind for ind in range(len(gallery_feat)-batch_last, len(gallery_feat))], gallery_feat[len(gallery_feat)-batch_last:,:]]
        collection.insert(data)
        res = count(cname)
        print('Number of entities: ', len(res))
    t2 = time.time()
    print('data has be inserted, cost time:', t2-t1)

错误信息如下：

-------省略很多Number of entities这些打印信息------
Number of entities: 1310000
Number of entities: 13200000
Number of entities: 13300000
Number of entities: 13400000
Number of entities: 13500000
Number of entities: 13600000
RPC error: [query], <BaseException: (code=1, message=err: rpc error: code = ResourceExhausted desc = trying to send message larger than max (105373137 vs. 104857600)
, /go/src/github.com/milvus-io/milvus/internal/util/trace/stack_trace.go:51 github.com/milvus-io/milvus/internal/util/trace.StackTrace
/go/src/github.com/milvus-io/milvus/internal/util/grpcclient/client.go:253 github.com/milvus-io/milvus/internal/util/grpcclient.(*ClientBase).ReCall
/go/src/github.com/milvus-io/milvus/internal/distributed/proxy/client/client.go:159 github.com/milvus-io/milvus/internal/distributed/proxy/client.(*Client).SendRetrieveResult
/go/src/github.com/milvus-io/milvus/internal/querynode/proxy_session_manager.go:120 github.com/milvus-io/milvus/internal/querynode.(*SessionManager).SendRetrieveResult
/go/src/github.com/milvus-io/milvus/internal/querynode/query_collection.go:1440 github.com/milvus-io/milvus/internal/querynode.(*queryCollection).publishRetrieveResultWithCtx
/go/src/github.com/milvus-io/milvus/internal/querynode/query_collection.go:1445 github.com/milvus-io/milvus/internal/querynode.(*queryCollection).publishRetrieveResult
/go/src/github.com/milvus-io/milvus/internal/querynode/query_collection.go:1359 github.com/milvus-io/milvus/internal/querynode.(*queryCollection).retrieve
/go/src/github.com/milvus-io/milvus/internal/querynode/query_collection.go:682 github.com/milvus-io/milvus/internal/querynode.(*queryCollection).doUnsolvedQueryMsg
/usr/local/go/src/runtime/asm_amd64.s:1371 runtime.goexit

)>, <Time:{'RPC start': '2022-06-10 14:37:44.837074', 'RPC error': '2022-06-10 14:37:51.441811'}>

[gobigrassland]

res = collection.query(
    expr = "idx>=0",
    output_fields = ["idx"],
    consistency_level="Strong"
    )

是不是collection.query这个函数但查询返回结果大于设定的gprc100M，就会报错？
idx是int64，占用8个字节。13600000*8/1024/1024=103.75M这个已经超过了100M，但是这个时候并没有报错。矛盾啊？

[yhmo]

query(expr = "idx>=0")相当于拿取所有的数据，随着insert插入越来越多的数据，query返回的数据也就越多。当数据超过100MB时，就会报trying to send message larger than max (105373137 vs. 104857600)的错误。

你在milvus.yaml里修改一下这两项，改大点试试：

grpc:
  clientMaxRecvSize: 104857600 # 100 MB, 100 * 1024 * 1024
  clientMaxSendSize: 104857600 # 100 MB, 100 * 1024 * 1024

[gobigrassland]

@yhmo 嗯嗯，是改yaml配置可以修改这个问题。但是随着collection的数量越来越多，通过query这种方法统计总的entities数量就不可行了（总会有超过grpc 512M限制的时候）。而采用colleciton.entities这种方法，当写入相同的数据，返回的collection.entities就会重复统计。

所以当collection中entities数量很多很多，业务中由于某些原因导致，存在很多重复写入的entities情况下，如何统计collection中reduced的不重复的entities数目呢？

[yhmo]

如果没有delete操作的话，用collection.num_entities就知道行数。
重复有两种意思：

1. id重复，目前还没有id查重功能，要等2.2或者2.3了。如果自己查重的话，用query(expr = "idx==xxx")，不过挺慢。
2. 向量重复，没什么好的办法，把向量丢进去搜一次，如果top1 的 L2 distance=0.0，就认为有重复。

[gobigrassland]

@yhmo 嗯呢，我说的重复是指id重复。当时测试时发现将相同idx相同的特征写入，entities数量会一直增加。我在https://github.com/milvus-io/milvus/discussions/17095这里问你们了。所以才找到query这种方法统计entities的数量。既然以后版本会支持，我暂时就不纠结这个问题了~