[Bug]: [streaming] After concurrent insertion and deletion, count(*) returns more results than expected

[ThreadDao]

Is there an existing issue for this?

• I have searched the existing issues

Environment

- Milvus version: master-20250228-13aff35a-amd64
- Deployment mode(standalone or cluster): cluster
- MQ type(rocksmq, pulsar or kafka):    
- SDK version(e.g. pymilvus v2.0.0rc2):
- OS(Ubuntu or CentOS): 
- CPU/Memory: 
- GPU: 
- Others:

Current Behavior

server

1 streaming node and 4 querynode

client

1. create a collection with pk (named id) field and vector field
2. create hnsw index for vector field
3. insert 30m entities, then flush
4. create index again, then load collection
5. concurrent requests:

• insert with batch 200, start_id 30,000,000 (Generate ids in order from 30m)
• delete from inserted ids with batch 150
• search

result

1. actual Count(*) results

c.query('', output_fields=["count(*)"])
data: ["{'count(*)': 47541450}"] , extra_info: {'cost': 0}

2.expected count(*): 78831000 - 229879 x 150= 44349150, It is 3,192,300 less than the actual

• From the last batch of insert debug logs we know that the maximum ID that has been inserted is 78830800 +200= 78831000
  
• From the delete debug log we know that there are 229879 delete requests

3. Even stronger evidence is from the birdwatcher results:

ster-insert-op-16-1453) > scan-binlog --collection 456316923026604608 --fields id --expr 0<=id --minioAddr 10.104.19.71 
=== Checking collection schema ===
PK Field [100] id
Output Field id field id 100
Start to connect to oss endpoind: 10.104.19.71:9000
Connection successful!
=== start to execute "count" task with filter expresion: "0<=id" ===
Total 44349150 entries found

Milvus(level-master-insert-op-16-1453) > scan-binlog --collection 456316923026604608 --fields id --expr 30000000<=id<78137550 --minioAddr 10.104.19.71
=== Checking collection schema ===
PK Field [100] id
Output Field id field id 100
Start to connect to oss endpoind: 10.104.19.71:9000
Connection successful!
=== start to execute "count" task with filter expresion: "30000000<=id<78137550" ===
Total 13655700 entries found 

The problem seems to be the delete part

Expected Behavior

No response

Steps To Reproduce

https://argo-workflows.zilliz.cc/archived-workflows/qa/86fdb7d3-267e-4447-a865-7632c4e1374a?nodeId=level-zero-stable-master-d-2071198988

Milvus Log

• client debug path: /test/fouram/log/2025_02_28/level-zero-stable-master-d-2_92192
• server pods

level-master-insert-op-16-1453-etcd-0                             1/1     Running                  0               3d      10.104.16.69    4am-node21   <none>           <none>
level-master-insert-op-16-1453-etcd-1                             1/1     Running                  0               3d      10.104.27.196   4am-node31   <none>           <none>
level-master-insert-op-16-1453-etcd-2                             1/1     Running                  0               3d      10.104.33.133   4am-node36   <none>           <none>
level-master-insert-op-16-1453-milvus-datanode-6cc767ffc6-mzwm6   1/1     Running                  0               3d      10.104.14.88    4am-node18   <none>           <none>
level-master-insert-op-16-1453-milvus-indexnode-558755c84d4kh95   1/1     Running                  0               3d      10.104.25.136   4am-node30   <none>           <none>
level-master-insert-op-16-1453-milvus-indexnode-558755c84d6scqn   1/1     Running                  0               3d      10.104.13.102   4am-node16   <none>           <none>
level-master-insert-op-16-1453-milvus-mixcoord-5d5fdcfff8-kbwbt   1/1     Running                  0               3d      10.104.17.174   4am-node23   <none>           <none>
level-master-insert-op-16-1453-milvus-proxy-7796d4b6dc-qtp4g      1/1     Running                  0               3d      10.104.13.101   4am-node16   <none>           <none>
level-master-insert-op-16-1453-milvus-querynode-0-78475d59gkh4d   1/1     Running                  0               3d      10.104.26.51    4am-node32   <none>           <none>
level-master-insert-op-16-1453-milvus-querynode-0-78475d59jg4kk   1/1     Running                  0               3d      10.104.23.191   4am-node27   <none>           <none>
level-master-insert-op-16-1453-milvus-querynode-0-78475d59l2rjt   1/1     Running                  0               3d      10.104.6.87     4am-node13   <none>           <none>
level-master-insert-op-16-1453-milvus-querynode-0-78475d59z6lkt   1/1     Running                  0               3d      10.104.27.203   4am-node31   <none>           <none>
level-master-insert-op-16-1453-milvus-streamingnode-6bf5d749qg4   1/1     Running                  0               3d      10.104.26.50    4am-node32   <none>           <none>
level-master-insert-op-16-1453-minio-0                            1/1     Running                  0               3d      10.104.19.71    4am-node28   <none>           <none>
level-master-insert-op-16-1453-minio-1                            1/1     Running                  0               3d      10.104.26.46    4am-node32   <none>           <none>
level-master-insert-op-16-1453-minio-2                            1/1     Running                  0               3d      10.104.27.197   4am-node31   <none>           <none>
level-master-insert-op-16-1453-minio-3                            1/1     Running                  0               3d      10.104.16.70    4am-node21   <none>           <none>

Anything else?

No response

[chyezh]

/assign

[dosubot]

I found a similar open issue regarding the count(*) query returning more results than expected after concurrent insertion and deletion in the Milvus streaming environment. The issue is related to duplicate Primary Keys (PKs) in different segments of your collection. When using count(*), Milvus counts duplicate PKs separately, while a regular query merges records with the same PK into one result. Additionally, Milvus uses soft deletes for upsert operations, which means that while the metadata (total_rows and indexed_rows) may show increasing numbers, the actual data count can be accurately retrieved using count(*) query with Strong consistency level. The old records aren't physically deleted immediately but will be cleaned up during compaction ^[1][2].

To continue talking to Dosu, mention @dosu.

---

^{Help Dosu learn! Give it feedback: Great Response | Irrelevant Answer | Incorrect Sources | Too Verbose | Hallucination | Bug Report | Other}

[chyezh]

There's no any data lost when consuming from streaming service.

DELETE:

INSERT:

Explain:

• There's a insert operation for build 30m entities, 7800 message is write into wal before loading, so there's 7800 offset for insert message.
• After loading, all consuming message count should be the double of append operation count.

Found the first deletion lost primary key range 31836000<=id<31836200.
which is not the multiple of delete batch size 150, but the multiple of insert batch size 200 .
If the delete message is lost at flush path, it should be the multiple of delete,
but it's multiple of insert, which may be lost at compaction path. some L1 segment is lost L0 apply.

still working on it.

[chyezh]

maybe related to #40207

[chyezh]

Ignore the previous related issue.
As issue described, the scan binlog tool confirms that there's no data loss at data path.

[chyezh]

@weiliu1031 please help to check if it is related to the enhancement #39552

[chyezh]

Found the range which is not deleted by delegator 44132450,44132500 by comparing client logs and current cluster query result.
44132450<=id<44132500 should be zero, but query result is 50.

The insert log at client side.

[2025-02-28 05:54:02,183 - DEBUG - fouram]: (api_request)  : [Collection.insert] args: <Collection.insert fields: 2, length: 200, content: [ [ `type<class 'int'>, dtype<>` 44132400 ... ], [ `type<class 'list'>, dtype<>` [0.5810249573103445, 0.8 ... ] ]>, [None], kwargs: {'timeout': 60}, [requestId: 69466dc2-f598-11ef-90ba-d674cc9560c9] (api_request.py:77)

The delete log is found at client side.

211048:[2025-02-28 06:01:45,079 - DEBUG - fouram]: (api_request)  : [Collection.delete] args: ['id in [44132450, 44132451, 44132452, 44132453, 44132454, 44132455, 44132456, 44132457, 44132458, 44132459, 44132460, 44132461, 44132462, 44132463, 44132464, 44132465, 44132466, 44132467, 44132468, 44132469, 44132470, 44132471, 44132472, 44132473, 44132474, 44132475, 44132476, 44132477, 44132478, 44132479, 44132480, 44132481, 44132482, 44132483, 44132484, 44132485, 44132486, 44132487, 44132488, 44132489, 44132490, 44132491, 44132492, 44132493, 44132494, 44132495, 44132496, 44132497, 44132498, 44132499, 44133000, 44133001, 44133002, 44133003, 44133004, 44133005, 44133006, 44133007, 44133008, 44133009, 44133010, 44133011, 44133012, 44133013, 44133014, 44133015, 44133016, 44133017, 44133018, 44133019, 44133020, 44133021, 44133022, 44133023, 44133024, 44133025, 44133026, 44133027, 44133028, 44133029, 44133030, 44133031, 44133032, 44133033, 44133034, 44133035, 44133036, 44133037, 44133038, 44133039, 44133040, 44133041, 44133042, 44133043, 44133044, 44133045, 44133046, 44133047, 44133048, 44133049, 44133050, 44133051, 44133052, 44133053, 44133054, 44133055, 44133056, 44133057, 44133058, 44133059, 44133060, 44133061, 44133062, 44133063, 44133064, 44133065, 44133066, 44133067, 44133068, 44133069, 44133070, 44133071, 44133072, 44133073, 44133074, 44133075, 44133076, 44133077, 44133078, 44133079, 44133080, 44133081, 44133082, 44133083, 44133084, 44133085, 44133086, 44133087, 44133088, 44133089, 44133090, 44133091, 44133092, 44133093, 44133094, 44133095, 44133096, 44133097, 44133098, 44133099]', None, 60], kwargs: {}, [requestId: 7d2ed382-f599-11ef-90ba-d674cc9560c9] (api_request.py:77)
211049-[2025-02-28 06:01:45,080 - DEBUG - fouram]: (api_response) : [Collection.delete] (insert count: 0, delete count: 150, upsert count: 0, timestamp: 456319959257579544, success count: 0, err count: 0, cost: 0), [requestId: 7ab82f0e-f599-11ef-90ba-d674cc9560c9] (api_request.py:44)
211050-[2025-02-28 06:01:45,080 - DEBUG - fouram]: [Time] Collection.delete run in 4.1333s (api_request.py:51)

The related delete timetick at server side.

2025-02-28 14:01:45.085	[2025/02/28 06:01:45.085 +00:00] [DEBUG] [metricsutil/wal_write.go:100] ["append message into wal"] [module=streamingnode] [component=wal] [channel="{\"Name\":\"level-master-insert-op-16-1453-rootcoord-dml_0\",\"Term\":2}"] [message_type=DELETE] [bytes=2206] [append_duration=2.825365ms] [impl_append_duration=2.79916ms] [presisted=true] [timetick_0="before: 9.315µs, after: 11.217µs, before_err: <nil>"] [segment-assign_0="before: 280ns, after: 214ns, before_err: <nil>"] [message_id=22/36611/0] [time_tick=456319960358584338]
2025-02-28 14:01:45.085	[2025/02/28 06:01:45.085 +00:00] [DEBUG] [producer/produce_server.go:229] ["send produce message response to client"] [module=streamingnode] [component=producer-server] [channel=level-master-insert-op-16-1453-rootcoord-dml_0] [term=2] [requestID=153456] [appendResult="{\"MessageID\":{\"MessageID\":{}},\"TimeTick\":456319960358584338,\"TxnCtx\":null,\"Extra\":null}"] []
2025-02-28 14:01:45.220	[2025/02/28 06:01:45.220 +00:00] [INFO] [writebuffer/l0_write_buffer.go:230] ["Add a new level zero segment"] [collectionID=456316923026604608] [channel=level-master-insert-op-16-1453-rootcoord-dml_0_456316923026604608v0] [segmentID=456316923074543954] [level=L0] ["start position"="channel_name:\"level-master-insert-op-16-1453-rootcoord-dml_0_456316923026604608v0\" msgID:\"\\x08\\x16\\x10\\x80\\x9e\\x02\\x18\\x00 \\x000\\x01\" timestamp:456319960358584338"]
2025-02-28 14:01:45.221	[2025/02/28 06:01:45.221 +00:00] [INFO] [pipeline/delete_node.go:56] ["pipeline fetch delete msg"] [collectionID=456316923026604608] [partitionID=-1] [deleteRowNum=150] [timestampMin=456319960358584338] [timestampMax=456319960358584338]
2025-02-28 14:01:45.221	[2025/02/28 06:01:45.221 +00:00] [DEBUG] [delegator/delegator_data.go:198] ["start to process delete"] [collectionID=456316923026604608] [channel=level-master-insert-op-16-1453-rootcoord-dml_0_456316923026604608v0] [replicaID=456316923449442305] [ts=456319960358584338]
2025-02-28 14:01:45.222	[2025/02/28 06:01:45.222 +00:00] [DEBUG] [querynodev2/services.go:1360] ["QueryNode received worker delete detail"] [collectionID=456316923026604608] [channel=level-master-insert-op-16-1453-rootcoord-dml_0_456316923026604608v0] [segmentID=456316923074543916] [scope=Streaming] [info="Pks range[44132450-44133099], len: 150, timestamp range: [456319960358584338-456319960358584338]"]
2025-02-28 14:01:45.240	[2025/02/28 06:01:45.240 +00:00] [DEBUG] [querynodev2/services.go:1360] ["QueryNode received worker delete detail"] [traceID=41032c826951f52325665d1ad023e6f7] [collectionID=456316923026604608] [channel=level-master-insert-op-16-1453-rootcoord-dml_0_456316923026604608v0] [segmentID=456316923075377269] [scope=Historical] [info="Pks range[44132450-44133099], len: 150, timestamp range: [456319960358584338-456319960358584338]"]
2025-02-28 14:01:46.338	[2025/02/28 06:01:46.338 +00:00] [INFO] [syncmgr/meta_writer.go:75] [SaveBinlogPath] [SegmentID=456316923074543654] [CollectionID=456316923026604608] [ParitionID=-1] [startPos="[{\"start_position\":{\"channel_name\":\"level-master-insert-op-16-1453-rootcoord-dml_0_456316923026604608v0\",\"msgID\":\"CBYQgJ4CGAAgADAB\",\"timestamp\":456319960358584338},\"segmentID\":456316923074543954}]"] [checkPoints="[{\"segmentID\":456316923074543654,\"position\":{\"channel_name\":\"level-master-insert-op-16-1453-rootcoord-dml_0_456316923026604608v0\",\"msgID\":\"CBYQgJ4CGAAgADAB\",\"timestamp\":456319960358584337}}]"] [binlogNum=0] [statslogNum=0] [deltalogNum=1] [bm25logNum=0] [vChannelName=level-master-insert-op-16-1453-rootcoord-dml_0_456316923026604608v0]

The delete operation is flushed in L0 segment 456316923074543954;

[2025/02/28 06:01:45.220 +00:00] [INFO] [writebuffer/l0_write_buffer.go:230] ["Add a new level zero segment"] [collectionID=456316923026604608] [channel=level-master-insert-op-16-1453-rootcoord-dml_0_456316923026604608v0] [segmentID=456316923074543954] [level=L0] ["start position"="channel_name:\"level-master-insert-op-16-1453-rootcoord-dml_0_456316923026604608v0\" msgID:\"\\x08\\x16\\x10\\x80\\x9e\\x02\\x18\\x00 \\x000\\x01\" timestamp:456319960358584338"]

The related insert segment at server side should be 456316923074543916, then stats into 456316923076206612.
Then compacted into 456316923076208782, then stats into 456316923076210994.
finally compacted into 456316923076627973.

[2025/02/28 06:03:58.974 +00:00] [INFO] [datacoord/meta.go:1686] ["Add a new compactTo segment"] [planID=456316923076208781] [type=MixCompaction] [collectionID=456316923026604608] [partitionID=456316923026604609] [channel=level-master-insert-op-16-1453-rootcoord-dml_0_456316923026604608v0] [compactFrom="[456316923075792509,456316923075800504,456316923076206612]"] [compactTo=456316923076208782] ["compactTo segment numRows"=1717550] ["binlog count"=4] ["statslog count"=1] ["deltalog count"=0]

Can be found from binlog and delete log at segment 456316923076627973

// binlog
120825:		120800 : 44132450
// deletelog
387025:		387000 : {"pk":44132450,"ts":456319960358584338,"pkType":5}

The load operation happens before L0 compaction:

[2025/02/28 06:13:15.328 +00:00] [INFO] [segments/segment_loader.go:1226] ["there are no delta logs saved with segment, skip loading delete record"] [traceID=c6a8ee8fb85260292671661adbbb4ca8] [segmentID=456316923076627973] [deltaNum=0]
...
[2025/02/28 06:13:15.495 +00:00] [INFO] [delegator/delegator_data.go:679] ["forward delete to worker..."] [traceID=c6a8ee8fb85260292671661adbbb4ca8] [collectionID=456316923026604608] [channel=level-master-insert-op-16-1453-rootcoord-dml_0_456316923026604608v0] [replicaID=456316923449442305] [segmentID=456316923076627973] [deleteRowNum=105750]
...
[2025/02/28 06:13:39.975 +00:00] [INFO] [datacoord/compaction_trigger_v2.go:373] ["Finish to submit a LevelZeroCompaction plan"] [view="L0SegCount=1, DeltaRowCount=617100, label=<coll=456316923026604608, part=-1, channel=level-master-insert-op-16-1453-rootcoord-dml_0_456316923026604608v0>, posT=<456320145432248343>, L0 segments=[<ID=456316923074543954, level=L0, deltaSize=30932138.00, deltaLogCount=1, deltaRowCount=617100>]"] [triggerID=456316923077044804] [planID=456316923077044804] [type=Level0DeleteCompaction] ["L0 segments"="[456316923074543954]"]

So the load operation lost some delete record at delete buffer, maybe lost a full L0 segment or with a wrong bloom filter?

[weiliu1031]

@weiliu1031 please help to check if it is related to the enhancement #39552

PR #39552 introduced a regression where the delete buffer's data is being prematurely cleaned up. This results in partial loss of necessary deletion data during subsequent load segment operations when processing forward deletes.

will be fixed by #40366

[weiliu1031]

/assign