Fix is_legal_ids() to accept negative int64 primary keys

pymilvus/client/check.py

@@ -1,5 +1,4 @@
 import datetime
-import sys
 from typing import Any, Callable, Union
 
 import numpy as np
@@ -11,6 +10,9 @@
 from . import entity_helper
 from .singleton_utils import Singleton
 
+_INT64_MIN = -(2**63)
+_INT64_MAX = 2**63 - 1
+
 
 def validate_strs(**kwargs):
     """validate if all values are legal non-emtpy str"""
@@ -130,7 +132,7 @@ def is_legal_ids(ids: Any) -> bool:
             if isinstance(i, bool) or not isinstance(i, (int, np.integer)):
                 return False
             value = int(i)
-            if value < 0 or value > sys.maxsize:
+            if not (_INT64_MIN <= value <= _INT64_MAX):
                 return False
         return True
 
@@ -142,7 +144,7 @@ def is_legal_ids(ids: Any) -> bool:
                 value = int(i)
             except (TypeError, ValueError, OverflowError):
                 continue
-            if value < 0 or value > sys.maxsize:
+            if not (_INT64_MIN <= value <= _INT64_MAX):
                 return False
         return True
 

tests/test_check.py

@@ -12,6 +12,7 @@
     check_pass_param,
     is_legal_address,
     is_legal_host,
+    is_legal_ids,
     is_legal_port,
 )
 from pymilvus.client.utils import (
@@ -86,6 +87,44 @@ def test_check_is_legal_port_false(self, invalid_port):
         assert valid is False
 
 
+class TestIsLegalIds:
+    @pytest.mark.parametrize(
+        "ids",
+        [
+            [1, 2, 3],
+            [0],
+            [-(2**63), 2**63 - 1],  # int64 min/max
+            [-1, -100, -9222883346732719253],  # negative int64 values
+            [np.int64(-1), np.int64(2**63 - 1)],
+        ],
+    )
+    def test_valid_int_ids(self, ids):
+        assert is_legal_ids(ids) is True
+
+    @pytest.mark.parametrize(
+        "ids",
+        [
+            ["abc", "def"],
+            ["-123", "456"],
+        ],
+    )
+    def test_valid_str_ids(self, ids):
+        assert is_legal_ids(ids) is True
+
+    @pytest.mark.parametrize(
+        "ids",
+        [
+            None,
+            [],
+            [True, False],
+            [2**63],  # exceeds int64 max
+            [-(2**63) - 1],  # exceeds int64 min
+        ],
+    )
+    def test_invalid_ids(self, ids):
+        assert is_legal_ids(ids) is False
+
+
 class TestCheckPassParam:
     def test_check_pass_param_valid(self):
         a = [[i * j for i in range(20)] for j in range(20)]