[Enhancement] allows to specufy temp local path for RemoteBulkWriter for temporarily persisting data files (#3077)

issue: #3015

Signed-off-by: yhmo <yihua.mo@zilliz.com>

Author: yhmo

examples/bulk_import/bulk_writer_all_types.py

@@ -216,6 +216,7 @@ def remote_writer(schema: CollectionSchema, file_type: BulkFileType):
     with RemoteBulkWriter(
             schema=schema,
             remote_path="bulk_data",
+            local_path="/tmp/PARQUET",
             connect_param=RemoteBulkWriter.S3ConnectParam(
                 endpoint=MINIO_ADDRESS,
                 access_key=MINIO_ACCESS_KEY,

pymilvus/bulk_writer/remote_bulk_writer.py

@@ -32,6 +32,8 @@
 
 logger = logging.getLogger(__name__)
 
+TEMP_LOCAL_PATH = "local_path"
+
 
 class RemoteBulkWriter(LocalBulkWriter):
     class S3ConnectParam:
@@ -112,8 +114,12 @@ def __init__(
         config: Optional[dict] = None,
         **kwargs,
     ):
-        local_path = Path(sys.argv[0]).resolve().parent.joinpath("bulk_writer")
-        super().__init__(schema, str(local_path), chunk_size, file_type, config, **kwargs)
+        temp_local_path = str(Path(sys.argv[0]).resolve().parent.joinpath("bulk_writer"))
+        if TEMP_LOCAL_PATH in kwargs:
+            temp_local_path = kwargs.get(TEMP_LOCAL_PATH)
+            kwargs.pop(TEMP_LOCAL_PATH)
+        super().__init__(schema, temp_local_path, chunk_size, file_type, config, **kwargs)
+
         self._remote_path = Path("/").joinpath(remote_path).joinpath(super().uuid)
         self._connect_param = connect_param
         self._client = None