How to get all the data of a collection?

[softhuafei]

I want to save all the data in a collection to csv or hdf5, but I can't find the corresponding query statement. Traditional databases can scan the entire table through select * from table, but the query statement of milvus must specify expr.

When I try to run query like this:

res = collection.query(
    expr = "1 == 1", 
  output_fields = ["book_id", "book_intro"],
  consistency_level="Strong"
)

it will throw error

---------------------------------------------------------------------------
MilvusException                           Traceback (most recent call last)
Input In [20], in <cell line: 1>()
----> 1 res = collection.query(
      2     expr = "1 == 1", 
      3   output_fields = ["book_id", "book_intro"],
      4   consistency_level="Strong"
      5 )

File ~/opt/anaconda3/lib/python3.9/site-packages/pymilvus/orm/collection.py:712, in Collection.query(self, expr, output_fields, partition_names, timeout, **kwargs)
    709     raise DataTypeNotMatchException(message=ExceptionsMessage.ExprType % type(expr))
    711 conn = self._get_connection()
--> 712 res = conn.query(self._name, expr, output_fields, partition_names,
    713                  timeout=timeout, schema=self._schema_dict, **kwargs)
    714 return res

File ~/opt/anaconda3/lib/python3.9/site-packages/pymilvus/decorators.py:109, in error_handler.<locals>.wrapper.<locals>.handler(*args, **kwargs)
    107     record_dict["RPC error"] = str(datetime.datetime.now())
    108     LOGGER.error(f"RPC error: [{inner_name}], {e}, <Time:{record_dict}>")
--> 109     raise e
    110 except grpc.FutureTimeoutError as e:
    111     record_dict["gRPC timeout"] = str(datetime.datetime.now())

File ~/opt/anaconda3/lib/python3.9/site-packages/pymilvus/decorators.py:105, in error_handler.<locals>.wrapper.<locals>.handler(*args, **kwargs)
    103 try:
    104     record_dict["RPC start"] = str(datetime.datetime.now())
--> 105     return func(*args, **kwargs)
    106 except MilvusException as e:
    107     record_dict["RPC error"] = str(datetime.datetime.now())

File ~/opt/anaconda3/lib/python3.9/site-packages/pymilvus/decorators.py:136, in tracing_request.<locals>.wrapper.<locals>.handler(self, *args, **kwargs)
    134 if req_id:
    135     self.set_onetime_request_id(req_id)
--> 136 ret = func(self, *args, **kwargs)
    137 return ret

File ~/opt/anaconda3/lib/python3.9/site-packages/pymilvus/decorators.py:85, in retry_on_rpc_failure.<locals>.wrapper.<locals>.handler(self, *args, **kwargs)
     83         back_off = min(back_off * back_off_multiplier, max_back_off)
     84     else:
---> 85         raise e
     86 except Exception as e:
     87     raise e

File ~/opt/anaconda3/lib/python3.9/site-packages/pymilvus/decorators.py:50, in retry_on_rpc_failure.<locals>.wrapper.<locals>.handler(self, *args, **kwargs)
     48 while True:
     49     try:
---> 50         return func(self, *args, **kwargs)
     51     except grpc.RpcError as e:
     52         # DEADLINE_EXCEEDED means that the task wat not completed
     53         # UNAVAILABLE means that the service is not reachable currently
     54         # Reference: https://grpc.github.io/grpc/python/grpc.html#grpc-status-code
     55         if e.code() != grpc.StatusCode.DEADLINE_EXCEEDED and e.code() != grpc.StatusCode.UNAVAILABLE:

File ~/opt/anaconda3/lib/python3.9/site-packages/pymilvus/client/grpc_handler.py:988, in GrpcHandler.query(self, collection_name, expr, output_fields, partition_names, timeout, **kwargs)
    986     return []
    987 if response.status.error_code != Status.SUCCESS:
--> 988     raise MilvusException(response.status.error_code, response.status.reason)
    990 num_fields = len(response.fields_data)
    991 # check has fields

MilvusException: <MilvusException: (code=1, message=fail to query on all shard leaders, err=All attempts results:
attempt #1:code: UnexpectedError, error: fail to Query, QueryNode ID = 26, reason=Query 26 failed, reason context canceled err %!w(<nil>)
attempt #2:context canceled
)>

[xiaofan-luan]

has to use iterator features on 2.3.
So far milvus has limitations on query size limit

[moonSandra]

Does this issue still exist? for version 2.5

[yhmo]

@moonSandra
In 2.5, also use iterator to fetch all the data.

An example to fetch all data of a collection to json files:

import random
import json

from pymilvus import (
    connections,
    FieldSchema, CollectionSchema, DataType,
    Collection,
    utility,
)

connections.connect(host='localhost', port=19530)
print(utility.get_server_version())

collection_name = "AAA"
dim = 128

fields = [
    FieldSchema(name="book_id", dtype=DataType.INT64, is_primary=True, auto_id=False),
    FieldSchema(name="vector", dtype=DataType.FLOAT_VECTOR, dim=dim),
    FieldSchema(name="book_intro", dtype=DataType.VARCHAR, max_length=65535),
]
schema = CollectionSchema(fields)
utility.drop_collection(collection_name)
collection = Collection(collection_name, schema)
print(collection_name, "created")

collection.create_index(field_name="vector", index_params={
        "metric_type": "L2",
        "index_type": "FLAT",
        "params": {}
    })
print("index created")
collection.load()

row_count = 10000
data = [
    [k for k in range(row_count)], # book_id
    [[random.random() for _ in range(dim)] for _ in range(row_count)], # vector
    [f"this is bool no.{k}" for k in range(row_count)], # book_intro
]
collection.insert(data=data)
print("data inserted")
res = collection.query(expr="", consistency_level="Strong", output_fields=["count(*)"])
print("count(*):", res[0]["count(*)"])

iterator = collection.query_iterator(
    batch_size=100,
    expr="",
    output_fields=["book_id", "book_intro"]
)

data = []
file_index = 0
def dump_data(file_name: str):
    with open(file_name, 'w') as f:
        json.dump(data, f)
        print(file_name, "saved")

while True:
    item = iterator.next()
    if not item:
        iterator.close()
        break

    data.extend(item)
    if len(data) >= 1000:
        dump_data(f'data_{file_index}.json')
        file_index = file_index + 1
        data = []

if len(data) >= 0:
    dump_data(f'data_{file_index}.json')

[moonSandra]

@yhmo

Thank you for providing this method.
I was thinking to another approach but i don't know this is a good idea or not. I was thinking about using pagination (offset and limit) for this issue. but i don't know milvus has limitation on this or not (for eg. elasticsearch cannot retrieve more than 10,000 for "from and size " method)

connections.connect(alias="conn", host="localhost", port="19530", db_name="Users", )
coll = Collection(name="sub_coll", using="conn")

page_size = 100
total_pages = (coll.num_entities//page_size) 

for page_number in range(1,total_pages):
    offset = (page_number - 1) * page_size
    results = coll.query(
        expr="", 
        output_fields=["*"],
        offset=offset,
        limit=page_size
    )
    results = list(results)

[yhmo]

In milvus, the max value of "limit" and "offset" is limited by 16384. (offset+limit) must be in range [1, 16384]. So, this approach doesn't work if your collection's row count exceeds 16384.