[Bug]: [Add Field]Add field cause original continuous insert operation to fail

[qixuan0212]

Is there an existing issue for this?

• [x] I have searched the existing issues

Environment

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

Current Behavior

1. continuous insert (no new field)
2. add new field
3. continuous insert fail

2025-04-30 15:51:27,358 [ERROR][handler]: RPC error: [insert_rows], <MilvusException: (code=1100, message=the length of valid_data of field(field_new_02) is wrong: invalid parameter[expected=10][actual=0])>, <Time:{'RPC start': '2025-04-30 15:51:27.342762', 'RPC error': '2025-04-30 15:51:27.358491'}> (decorators.py:140)

Expected Behavior

Add field will not cause original continuous insert operation to fail

Steps To Reproduce

1. continuous insert (no new field)
2. add new field
3. continuous insert fail

Milvus Log

https://grafana-4am.zilliz.cc/explore?orgId=1&panes=%7B%22SHh%22:%7B%22datasource%22:%22vhI6Vw67k%22,%22queries%22:%5B%7B%22refId%22:%22A%22,%22expr%22:%22%7Bcluster%3D%5C%22devops%5C%22,namespace%3D%5C%22chaos-testing%5C%22,pod%3D~%5C%22rerank-master-1-uvyhi.%2A%5C%22%7D%22,%22datasource%22:%7B%22type%22:%22loki%22,%22uid%22:%22vhI6Vw67k%22%7D%7D%5D,%22range%22:%7B%22from%22:%221745999473125%22,%22to%22:%221745999501888%22%7D%7D%7D&schemaVersion=1

Anything else?

def insert_loop(thread_id):
    global id_counter
    while True:
        with id_lock:
            start_id = id_counter
            id_counter += nb_per_batch

        vectors = np.random.rand(nb_per_batch, dim).astype(np.float32)
        data = [{
            "my_id": start_id + i,
            "my_vector": vectors[i],
            "my_varchar": f"varchar_thread{thread_id}_{start_id+i}"
        } for i in range(nb_per_batch)]

        try:
            client.insert(collection_name=collection_name, data=data)
            print(f"[Thread-{thread_id}] Inserted IDs {start_id} to {start_id + nb_per_batch - 1}")
        except Exception as e:
            print(f"[Thread-{thread_id}] Insert error: {e}")

        time.sleep(5)

[yanliang567]

/assign @congqixia /unassign

[qixuan0212]

Retest version: master-20250511-52950ce3-amd64 pymilvus:2.6.0rc119

result: streamingnode crash

log: https://grafana-4am.zilliz.cc/explore?orgId=1&panes=%7B%22-sJ%22:%7B%22datasource%22:%22vhI6Vw67k%22,%22queries%22:%5B%7B%22refId%22:%22A%22,%22expr%22:%22%7Bcluster%3D%5C%22devops%5C%22,namespace%3D%5C%22chaos-testing%5C%22,pod%3D~%5C%22qx4-master-addfield-gdneh.%2A%5C%22%7D%22,%22datasource%22:%7B%22type%22:%22loki%22,%22uid%22:%22vhI6Vw67k%22%7D,%22editorMode%22:%22code%22,%22queryType%22:%22range%22%7D%5D,%22range%22:%7B%22from%22:%22now-5m%22,%22to%22:%22now%22%7D%7D%7D&schemaVersion=1

[qixuan0212]

crash issue，so set it as critical.

[qixuan0212]

Verification passed

milvus: master-20250519-38ded736-amd64 pymilvus: 2.6.0rc122