feat: support force l0 compact(#3008)

pymilvus/client/async_grpc_handler.py

@@ -1825,6 +1825,7 @@ async def compact(
         self,
         collection_name: str,
         is_clustering: Optional[bool] = False,
+        is_l0: Optional[bool] = False,
         timeout: Optional[float] = None,
         **kwargs,
     ) -> int:
@@ -1835,7 +1836,9 @@ async def compact(
         )
         check_status(response.status)
 
-        req = Prepare.manual_compaction(collection_name, is_clustering, response.collectionID)
+        req = Prepare.manual_compaction(
+            collection_name, is_clustering, is_l0, response.collectionID
+        )
         response = await self._async_stub.ManualCompaction(req, timeout=timeout, metadata=meta)
         check_status(response.status)
 

pymilvus/client/grpc_handler.py

@@ -1776,12 +1776,13 @@ def compact(
         self,
         collection_name: str,
         is_clustering: Optional[bool] = False,
+        is_l0: Optional[bool] = False,
         timeout: Optional[float] = None,
         **kwargs,
     ) -> int:
         meta = _api_level_md(**kwargs)
         # try with only collection_name
-        req = Prepare.manual_compaction(collection_name, is_clustering)
+        req = Prepare.manual_compaction(collection_name, is_clustering, is_l0)
         response = self._stub.ManualCompaction(req, timeout=timeout, metadata=meta)
         if response.status.error_code == common_pb2.CollectionNameNotFound:
             # should be removed, but to be compatible with old milvus server, keep it for now.

pymilvus/client/prepare.py

@@ -1567,17 +1567,23 @@ def load_balance_request(
 
     @classmethod
     def manual_compaction(
-        cls, collection_name: str, is_clustering: bool, collection_id: Optional[int] = None
+        cls,
+        collection_name: str,
+        is_clustering: bool,
+        is_l0: bool,
+        collection_id: Optional[int] = None,
     ):
         if is_clustering is None or not isinstance(is_clustering, bool):
             raise ParamError(message=f"is_clustering value {is_clustering} is illegal")
+        if is_l0 is None or not isinstance(is_l0, bool):
+            raise ParamError(message=f"is_l0 value {is_l0} is illegal")
 
         request = milvus_types.ManualCompactionRequest()
         if collection_id is not None:
             request.collectionID = collection_id
         request.collection_name = collection_name
         request.majorCompaction = is_clustering
-
+        request.l0Compaction = is_l0
         return request
 
     @classmethod

pymilvus/grpc_gen/milvus_pb2.pyi

@@ -1390,7 +1390,7 @@ class LoadBalanceRequest(_message.Message):
     def __init__(self, base: _Optional[_Union[_common_pb2.MsgBase, _Mapping]] = ..., src_nodeID: _Optional[int] = ..., dst_nodeIDs: _Optional[_Iterable[int]] = ..., sealed_segmentIDs: _Optional[_Iterable[int]] = ..., collectionName: _Optional[str] = ..., db_name: _Optional[str] = ...) -> None: ...
 
 class ManualCompactionRequest(_message.Message):
-    __slots__ = ("collectionID", "timetravel", "majorCompaction", "collection_name", "db_name", "partition_id", "channel", "segment_ids")
+    __slots__ = ("collectionID", "timetravel", "majorCompaction", "collection_name", "db_name", "partition_id", "channel", "segment_ids", "l0Compaction")
     COLLECTIONID_FIELD_NUMBER: _ClassVar[int]
     TIMETRAVEL_FIELD_NUMBER: _ClassVar[int]
     MAJORCOMPACTION_FIELD_NUMBER: _ClassVar[int]
@@ -1399,6 +1399,7 @@ class ManualCompactionRequest(_message.Message):
     PARTITION_ID_FIELD_NUMBER: _ClassVar[int]
     CHANNEL_FIELD_NUMBER: _ClassVar[int]
     SEGMENT_IDS_FIELD_NUMBER: _ClassVar[int]
+    L0COMPACTION_FIELD_NUMBER: _ClassVar[int]
     collectionID: int
     timetravel: int
     majorCompaction: bool
@@ -1407,7 +1408,8 @@ class ManualCompactionRequest(_message.Message):
     partition_id: int
     channel: str
     segment_ids: _containers.RepeatedScalarFieldContainer[int]
-    def __init__(self, collectionID: _Optional[int] = ..., timetravel: _Optional[int] = ..., majorCompaction: bool = ..., collection_name: _Optional[str] = ..., db_name: _Optional[str] = ..., partition_id: _Optional[int] = ..., channel: _Optional[str] = ..., segment_ids: _Optional[_Iterable[int]] = ...) -> None: ...
+    l0Compaction: bool
+    def __init__(self, collectionID: _Optional[int] = ..., timetravel: _Optional[int] = ..., majorCompaction: bool = ..., collection_name: _Optional[str] = ..., db_name: _Optional[str] = ..., partition_id: _Optional[int] = ..., channel: _Optional[str] = ..., segment_ids: _Optional[_Iterable[int]] = ..., l0Compaction: bool = ...) -> None: ...
 
 class ManualCompactionResponse(_message.Message):
     __slots__ = ("status", "compactionID", "compactionPlanCount")