gemini-code-assist[bot] commented on code in PR #39378:
URL: https://github.com/apache/beam/pull/39378#discussion_r3607576403
##########
sdks/python/apache_beam/ml/rag/ingestion/milvus_search.py:
##########
@@ -287,16 +287,18 @@ def write(self, documents):
"""Writes a batch of documents to the Milvus collection.
Performs an upsert operation to insert new documents or update existing
- ones based on primary key. After the upsert, flushes the collection to
- ensure data persistence.
+ ones based on primary key, reusing the client established in ``__enter__``.
+
+ Note:
+ This submits the upsert to Milvus; it does not call ``flush()`` on the
+ collection. Newly written rows become queryable according to Milvus'
+ normal segment-sync behavior. Callers that require immediate
+ read-after-write visibility should flush the collection explicitly.
Args:
documents: List of dictionaries representing Milvus records to write.
Each dictionary should contain fields matching the collection schema.
"""
- self._client = MilvusClient(
- **unpack_dataclass_with_kwargs(self._connection_params))
-
resp = self._client.upsert(
Review Comment:
To adhere to defensive programming practices and optimize efficiency, we
should add guard clauses to handle cases where `documents` is empty or
`self._client` is not initialized (e.g., if the sink is used outside of its
context manager lifecycle). This avoids unnecessary API calls and prevents an
unclear `AttributeError` by raising a descriptive `RuntimeError` instead.
```suggestion
if not documents:
return
if not getattr(self, '_client', None):
raise RuntimeError(
"Milvus client is not initialized. Please use this sink within a
context manager.")
resp = self._client.upsert(
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
