Can the insert or upsert operation on the client side ensure atomicity?

[gaozh424]

Hi ，
In my scenario, I inserted 70 million data entries into a collection in the Milvus cluster using the upsert method from the Java client. I then checked if the insertion was successful using getStatus(). My cluster is built on an NFS file system, and during the insertion process, the Minio component encountered several errors, likely due to insufficient performance of the NFS, which were resolved by restarting the Minio component. After the data synchronization was completed, I found that only 30 million data entries were present in Milvus, despite receiving a success message for 70 million entries.
Now, I can only re-synchronize all the data but cannot guarantee that the same issue won't occur during the synchronization process. What should I do? As far as I understand, isn't the data temporarily stored in Pulsar? Can Pulsar not guarantee data consistency?

Looking forward to your assistance.

[xiaofan-luan]

where did you store your data in pulsar?
If milvus returned write success, then data should be persistent.
Possibilities:

1. duplicated primary key
2. meta is not accuret, try to use count(*) and check how many entities is in this collection

[gaozh424]

Thank you for your reply.
I did not store the data in Pulsar.
As I understand it, Milvus first writes the data to Pulsar, and then the dataNode persists the data to MinIO. Am I mistaken?
You mentioned that the metadata is inaccurate; how can I confirm this?
Where can I execute a count(*)?
Sorry, I am not very familiar with Milvus.

[xiaofan-luan]

You r right.
Data is first persisted in pulsar/kafka, and later convert to minio.
you can execute count(*) by issue a query request , see https://milvus.io/docs/get-and-scalar-query.md

[gaozh424]

Is this feature only supported in version 2.4? I am currently using version 2.3.10, and the count(*) query returns empty.

[yhmo]

When you insert some data by insert(), what happens inside the Milvus:

1. The data is received by proxy node, proxy node splits the data into pieces by shards_number(which is defined in collection schema, default is 1), and sends the pieces(as messages) into Pulsar/Kafka. The insert() returns success to the client once the data is persisted into Pulsar/Kafka.
2. For each growing segment, two "data channels" are created, one is between querynode and Pulsar/Kafka, the other is between datanode and Pulsar/Kafka.
3. Querynode and datanode subscribe messages asynchronously from Pulsar/Kafka, the data pieces are consumed by querynode and datanode.
4. For each growing segment, datanode keeps a buffer to accept new data. Once the buffer size reaches 16MB or the buffer lifetime hits the limit(10 minutes), the buffer is persisted to be a chunk into S3/MinIO.
5. If the collection is loaded, all segments are kept in query nodes. for each growing segment, querynode keeps a buffer to accept new data, so that search requests can search the data. New data is consumed by querynode from Pulsar/Kafka, only when the new data is consumed by querynode, the data is visible&searchable by search()/query().
6. With more and more chunks persisted, the size of the growing segment reaches a threshold(100+MB), the growing segment is converted to be sealed segment by data coordinator. No more data is allowed to be inserted into this segment. Query node and data node will create a new buffer for the next growing segment.
7. Once a new sealed segment is generated, index node will receive a task to build index for this seal segment.
8. Once an index is finished, the query node will load this index into memory to replace the original segment data.

[gaozh424]

Thank you for the detailed and patient explanation.
I haven't pinpointed the exact issue yet. During the data insertion process, the disk space occupied by the Pulsar Bookie component keeps growing. I have set the parameters for Pulsar using bin/pulsar-admin namespaces set-retention public/default --size 20G --time 1d. According to the Pulsar documentation, this parameter will only take effect on messages that have already been consumed. I am unsure if this change will affect the DataNode.

[yhmo]

Set retention time to 1d is ok. But no need to set size limit to 20G.

Normally, datanode/querynode consumes the data very quickly if you don't have continually/intensive/huge data inserted.
In edge cases, for example, user uses multiple clients to insert data row by row, with thousands of insert actions per second. Datanode will more and more delay to consume the data, finally some data is expired before consumed into datanode, then the data will be lost.

[yhmo]

In milvus, data visible is controlled by the Consistency level: https://milvus.io/docs/consistency.md#Consistency

You insert some data, you want to know when the data can be searched. The correct way is to set the consistency level.
For query():
collection.query(consistency_level="Strong", expr="xxxx")

For search():
collection.search(consistency_level="Strong", data=vectors, anns_field="xxx")

[gaozh424]

On the Pulsar official website, I came across this statement: "The retention settings apply to all messages on topics that do not have any subscriptions, or to messages that have been acknowledged by all subscriptions. The retention policy settings do not affect unacknowledged messages on topics with subscriptions. The unacknowledged messages are controlled by the backlog quota.", therefore, I believe that the retention settings should not affect the normal operation of dataNodes, as the deletion pertains to the "historical messages" that have already been processed and subscribed to by the dataNode. Is my understanding correct? If not, this could be the key issue. My current scenario involves a large amount of continuous data insertion, as it is in the initial stage of setting up the milvus cluster, and since it is a cloud environment with limited resources provided by the client, the retention parameters of Pulsar have been restricted.

[yhmo]

Yes, seems the retention configurations only affect unacknowledged messages. Then 1d/20G makes sense.
Once a message is consumed by query nodes and data nodes, it will be marked as acknowledged and waiting for retention.