test: give each RemoteBulkWriter a private local_path (#54272)

## Problem

Without `local_path`, `RemoteBulkWriter` stages files under
`<dirname(sys.argv[0])>/bulk_writer`, so every pytest-xdist worker
shares one parent directory. pymilvus creates that parent and the
per-writer uuid subdirectory in two separate `mkdir` calls, and
`__exit__` removes the parent when it looks empty. When one worker exits
between another worker's two `mkdir` calls, the second `mkdir` fails
with `FileNotFoundError`.

Observed as `test_import_without_embedding[False-numpy]` failing under
`-n 32`.

## Fix

Pass `local_path=tempfile.mkdtemp(prefix="bulk_writer_")` at every
`RemoteBulkWriter` call site in `tests/python_client` so no two writers
share a parent directory. pymilvus still removes the empty directory on
exit, so nothing is left behind.

Test-only change; no product code touched.

## Verification

Ran the full PR-tier e2e (`ci_e2e_4am.sh`, L0+L1, `-n 32`) on a
standalone built from this branch: 9113 passed / 428 skipped, including
all 6 `test_import_without_embedding` parametrizations and all
bulk_insert / data_integrity cases. The only failures were 7
`test_milvus_client_pyudf` cases caused by the PyUDF runtime module not
being installed on the test host, unrelated to this change.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Signed-off-by: zhenshan.cao <zhenshan.cao@zilliz.com>
Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>

tests/python_client/bulk_insert/test_bulk_insert_api.py

@@ -1,6 +1,7 @@
 import hashlib
 import logging
 import random
+import tempfile
 import time
 from pathlib import Path
 
@@ -171,6 +172,7 @@ class TestBulkInsert(TestcaseBaseBulkInsert):
         self.collection_wrap.init_collection(c_name, schema=schema)
 
         with RemoteBulkWriter(
+            local_path=tempfile.mkdtemp(prefix="bulk_writer_"),
             schema=schema,
             remote_path=f"bulk_data/{c_name}/{file_type.name.lower()}",
             connect_param=RemoteBulkWriter.ConnectParam(

tests/python_client/chaos/checker.py

@@ -4,6 +4,7 @@ import math
 import os
 import random
 import re
+import tempfile
 import threading
 import time
 import unittest
@@ -3251,6 +3252,7 @@ class BulkInsertChecker(Checker):
 
     def prepare(self, data_size=100000):
         with RemoteBulkWriter(
+            local_path=tempfile.mkdtemp(prefix="bulk_writer_"),
             schema=self.schema,
             file_type=BulkFileType.NUMPY,
             remote_path="bulk_data",
@@ -3545,6 +3547,7 @@ class Import2PCChecker(Checker):
 
     def _write_import_files(self, rows):
         with RemoteBulkWriter(
+            local_path=tempfile.mkdtemp(prefix="bulk_writer_"),
             schema=self.schema,
             file_type=BulkFileType.PARQUET,
             remote_path=self.import_remote_path,

tests/python_client/milvus_client/test_milvus_client_continuous_data_integrity.py

@@ -1,6 +1,7 @@
 import json
 import os
 import queue
+import tempfile
 import threading
 import time
 from concurrent.futures import ThreadPoolExecutor
@@ -648,6 +649,7 @@ class TestMilvusClientContinuousStreamingDataIntegrity(TestMilvusClientV2Base):
                 sequence_start = _continuous_lane_sequence(epoch, 2, 0)
                 sequence_end = _continuous_lane_sequence(epoch, 2, CONTINUOUS_IMPORT_ROWS_PER_EPOCH - 1)
                 with RemoteBulkWriter(
+                    local_path=tempfile.mkdtemp(prefix="bulk_writer_"),
                     schema=schema,
                     remote_path=f"continuous_integrity/{collection_name}/epoch_{epoch + 1}",
                     connect_param=RemoteBulkWriter.ConnectParam(

tests/python_client/milvus_client/test_milvus_client_data_integrity.py

@@ -4,6 +4,7 @@ import math
 import os
 import random
 import struct
+import tempfile
 import time
 from contextlib import contextmanager
 
@@ -3883,6 +3884,7 @@ class TestMilvusClientCompactionDataIntegrity(TestMilvusClientV2Base):
             primary_key_type,
         )
         with RemoteBulkWriter(
+            local_path=tempfile.mkdtemp(prefix="bulk_writer_"),
             schema=schema,
             remote_path=f"compaction_integrity/{collection_name}/round_{round_index + 1}",
             connect_param=RemoteBulkWriter.ConnectParam(

tests/python_client/testcases/test_bulk_insert.py

@@ -1,6 +1,7 @@
 import json
 import logging
 import random
+import tempfile
 import time
 import uuid
 from pathlib import Path
@@ -442,6 +443,7 @@ class TestBulkInsertNullableVector(TestcaseBaseBulkInsert):
         null_ids = []
         non_null_ids = []
         with RemoteBulkWriter(
+            local_path=tempfile.mkdtemp(prefix="bulk_writer_"),
             schema=schema,
             remote_path="bulk_data",
             connect_param=RemoteBulkWriter.ConnectParam(
@@ -2161,6 +2163,7 @@ class TestBulkInsert(TestcaseBaseBulkInsert):
         schema = cf.gen_collection_schema(fields=fields, auto_id=auto_id, enable_dynamic_field=enable_dynamic_field)
         self.collection_wrap.init_collection(c_name, schema=schema)
         with RemoteBulkWriter(
+            local_path=tempfile.mkdtemp(prefix="bulk_writer_"),
             schema=schema,
             remote_path="bulk_data",
             connect_param=RemoteBulkWriter.ConnectParam(
@@ -2330,6 +2333,7 @@ class TestBulkInsert(TestcaseBaseBulkInsert):
         else:
             raise Exception(f"not support file format:{file_format}")
         with RemoteBulkWriter(
+            local_path=tempfile.mkdtemp(prefix="bulk_writer_"),
             schema=schema,
             remote_path="bulk_data",
             connect_param=RemoteBulkWriter.ConnectParam(
@@ -2469,6 +2473,7 @@ class TestBulkInsert(TestcaseBaseBulkInsert):
         schema = cf.gen_collection_schema(fields=fields, auto_id=auto_id, enable_dynamic_field=enable_dynamic_field)
         self.collection_wrap.init_collection(c_name, schema=schema)
         with RemoteBulkWriter(
+            local_path=tempfile.mkdtemp(prefix="bulk_writer_"),
             schema=schema,
             remote_path="bulk_data",
             connect_param=RemoteBulkWriter.ConnectParam(
@@ -2611,6 +2616,7 @@ class TestBulkInsert(TestcaseBaseBulkInsert):
         schema = cf.gen_collection_schema(fields=fields, auto_id=auto_id, enable_dynamic_field=enable_dynamic_field)
         self.collection_wrap.init_collection(c_name, schema=schema)
         with RemoteBulkWriter(
+            local_path=tempfile.mkdtemp(prefix="bulk_writer_"),
             schema=schema,
             remote_path="bulk_data",
             connect_param=RemoteBulkWriter.ConnectParam(
@@ -2766,6 +2772,7 @@ class TestBulkInsert(TestcaseBaseBulkInsert):
         schema = cf.gen_collection_schema(fields=fields, auto_id=auto_id, enable_dynamic_field=enable_dynamic_field)
         self.collection_wrap.init_collection(c_name, schema=schema)
         with RemoteBulkWriter(
+            local_path=tempfile.mkdtemp(prefix="bulk_writer_"),
             schema=schema,
             remote_path="bulk_data",
             connect_param=RemoteBulkWriter.ConnectParam(
@@ -3316,6 +3323,7 @@ class TestImportWithTextEmbeddingFunction(TestcaseBase):
         else:
             file_type = BulkFileType.PARQUET
         with RemoteBulkWriter(
+            local_path=tempfile.mkdtemp(prefix="bulk_writer_"),
             schema=schema,
             remote_path="bulk_data",
             connect_param=RemoteBulkWriter.ConnectParam(
@@ -3414,6 +3422,7 @@ class TestImportWithFunctionNegative(TestcaseBase):
         else:
             file_type = BulkFileType.PARQUET
         with RemoteBulkWriter(
+            local_path=tempfile.mkdtemp(prefix="bulk_writer_"),
             schema=invalid_schema,
             remote_path="bulk_data",
             connect_param=RemoteBulkWriter.ConnectParam(
@@ -3487,6 +3496,7 @@ class TestImportWithFunctionNegative(TestcaseBase):
         else:
             file_type = BulkFileType.PARQUET
         with RemoteBulkWriter(
+            local_path=tempfile.mkdtemp(prefix="bulk_writer_"),
             schema=invalid_schema,
             remote_path="bulk_data",
             connect_param=RemoteBulkWriter.ConnectParam(