[NA] [SDK] fix: address review comments on dataset insert deduplication

sdks/python/src/opik/api_objects/dataset/dataset.py

@@ -353,6 +353,11 @@ def __init__(
         # one-shot sync on the first `insert()` instead of paying an N+1
         # sync at list time.
         self._hashes_synced: bool = True
+        # None until the backend version has actually been determined. Only a
+        # conclusive answer is stored, so a probe that failed to reach the
+        # backend is retried instead of pinning this dataset to sequential
+        # uploads for the rest of the session.
+        self._parallel_insert_supported_cache: Optional[bool] = None
 
     @classmethod
     def from_public(
@@ -611,44 +616,60 @@ def _insert_batch_with_retry(
         )
         LOGGER.debug("Successfully sent dataset items batch of size %d", len(batch))
 
-    @functools.cached_property
+    @property
     def _parallel_insert_supported(self) -> bool:
         """Whether the backend tolerates concurrent batches sharing a batch_group_id.
 
         Older backends race on them, so parallelism is only safe from
         ``constants.MIN_BACKEND_VERSION_FOR_PARALLEL_INSERT`` onwards. When the
-        version cannot be determined at all — unreachable endpoint, non-semver
-        build string — we report unsupported rather than risk the race.
-
-        Cached because the backend cannot change version mid-session, and
-        parallel upload is the default: probing per ``insert`` would add a
-        round trip to every call in a loop.
+        version cannot be determined we report unsupported rather than risk the
+        race.
+
+        The answer is cached because the backend cannot change version
+        mid-session and parallel upload is the default: probing per ``insert``
+        would add a round trip to every call in a loop. Only a conclusive
+        answer is cached — an unreachable backend is re-probed on the next
+        insert so parallel upload resumes once it recovers.
         """
+        if self._parallel_insert_supported_cache is not None:
+            return self._parallel_insert_supported_cache
+
         try:
             backend_version = self._rest_client.version()["version"]
+        except Exception:
+            LOGGER.warning(
+                "Could not reach the Opik backend to determine its version, "
+                "falling back to a sequential dataset upload for this insert.",
+                exc_info=True,
+            )
+            return False
+
+        try:
             supported = (
                 semantic_version.SemanticVersion.parse(backend_version)
                 >= constants.MIN_BACKEND_VERSION_FOR_PARALLEL_INSERT
             )
         except Exception:
             LOGGER.warning(
-                "Could not determine the Opik backend version, falling back to a "
+                "Could not parse the Opik backend version %s, falling back to a "
                 "sequential dataset upload. Parallel upload requires backend %s "
                 "or newer.",
-                constants.MIN_BACKEND_VERSION_FOR_PARALLEL_INSERT,
-                exc_info=True,
-            )
-            return False
-
-        if not supported:
-            LOGGER.warning(
-                "Opik backend %s does not support parallel dataset upload, falling "
-                "back to a sequential upload. Upgrade to backend %s or newer to use "
-                "num_threads.",
                 backend_version,
                 constants.MIN_BACKEND_VERSION_FOR_PARALLEL_INSERT,
+                exc_info=True,
             )
+            supported = False
+        else:
+            if not supported:
+                LOGGER.warning(
+                    "Opik backend %s does not support parallel dataset upload, "
+                    "falling back to a sequential upload. Upgrade to backend %s or "
+                    "newer to use num_threads.",
+                    backend_version,
+                    constants.MIN_BACKEND_VERSION_FOR_PARALLEL_INSERT,
+                )
 
+        self._parallel_insert_supported_cache = supported
         return supported
 
     def _send_batches(
@@ -718,6 +739,12 @@ def __internal_api__insert_items_as_dataclasses__(
         num_threads: int = 1,
         deduplication: bool = True,
     ) -> None:
+        # Validated here rather than in each public entry point: every insert
+        # path funnels through this method. A truthy string or None would
+        # otherwise silently pick the wrong duplicate-checking behaviour.
+        if not isinstance(deduplication, bool):
+            raise ValueError("deduplication must be a bool")
+
         if deduplication:
             items_to_send = self._deduplicate(items)
         else:
@@ -756,7 +783,8 @@ def insert(
             deduplication: Whether to skip items whose content already exists
                 in the dataset. Pass ``False`` to insert every item as-is
                 without any duplicate checking, which is significantly faster
-                on large datasets.
+                on large datasets. The next insert that does deduplicate has to
+                re-read the dataset's items to account for what was skipped.
             num_threads: Number of worker threads used to upload the item
                 batches. Must be a positive integer, defaults to ``4``; pass
                 ``1`` to upload sequentially. All batches land in a single
@@ -769,6 +797,9 @@ def insert(
             raise ValueError("num_threads must be a positive integer")
         if num_threads < 1:
             raise ValueError("num_threads must be a positive integer")
+        # Checked before the version probe below so bad input costs no request.
+        if not isinstance(deduplication, bool):
+            raise ValueError("deduplication must be a bool")
 
         if num_threads > 1 and not self._parallel_insert_supported:
             num_threads = 1

sdks/python/src/opik/api_objects/dataset/rest_operations.py

@@ -284,6 +284,10 @@ def get_test_suites(
                 dataset_items_count=dataset_fern.dataset_items_count,
                 client=client,
             )
+            # This suite already holds items on the backend that we have not
+            # hashed locally, so the first insert must sync before it can tell
+            # a duplicate from a new item.
+            suite_dataset.__internal_api__hashes_synced__ = False
 
             suites.append(
                 test_suite_module.TestSuite(

sdks/python/tests/unit/api_objects/dataset/test_dataset_client.py

@@ -551,6 +551,68 @@ def test_insert__version_endpoint_unreachable__uploads_sequentially(monkeypatch)
     ), "A failing version probe must not break insert; it falls back to sequential"
 
 
+def test_insert__version_probe_recovers__parallel_upload_resumes(monkeypatch):
+    """A transient probe failure must not pin the dataset to sequential uploads."""
+    _small_batches(monkeypatch, size=_GATE_BATCH_SIZE)
+    mock_rest_client = Mock()
+    mock_rest_client.version.side_effect = [
+        ConnectionError("backend unreachable"),
+        {"version": constants.MIN_BACKEND_VERSION_FOR_PARALLEL_INSERT},
+    ]
+    dataset = Dataset(
+        name="test_dataset",
+        description="Test description",
+        project_name="Test project",
+        rest_client=mock_rest_client,
+    )
+
+    dataset.insert(_make_items(4), deduplication=False)
+    dataset.insert(_make_items(4), deduplication=False)
+
+    assert mock_rest_client.version.call_count == 2, (
+        "The failed probe must be retried rather than cached as unsupported"
+    )
+    assert dataset._parallel_insert_supported, (
+        "Once the backend answers, parallel upload must be available again"
+    )
+
+
+def test_insert__unparseable_version__probed_once(monkeypatch):
+    """An unparseable version is a conclusive answer, so it must still cache."""
+    _small_batches(monkeypatch, size=_GATE_BATCH_SIZE)
+    mock_rest_client = _mock_rest_client("dev-local")
+    dataset = Dataset(
+        name="test_dataset",
+        description="Test description",
+        project_name="Test project",
+        rest_client=mock_rest_client,
+    )
+
+    for _ in range(3):
+        dataset.insert(_make_items(4), deduplication=False)
+
+    assert mock_rest_client.version.call_count == 1, (
+        "A version the SDK cannot parse will not change, so it must not be re-probed"
+    )
+
+
+@pytest.mark.parametrize("bad_value", ["false", 0, 1, None, "", []])
+def test_insert__non_bool_deduplication__raises_before_any_request(bad_value):
+    mock_rest_client = Mock()
+    dataset = Dataset(
+        name="test_dataset",
+        description="Test description",
+        project_name="Test project",
+        rest_client=mock_rest_client,
+    )
+
+    with pytest.raises(ValueError, match="deduplication must be a bool"):
+        dataset.insert(_make_items(3), deduplication=bad_value)
+
+    mock_rest_client.datasets.create_or_update_dataset_items.assert_not_called()
+    mock_rest_client.version.assert_not_called()
+
+
 def test_insert__sequential__uploads_sequentially_without_probing_version(monkeypatch):
     mock_rest_client = _mock_rest_client()
 

sdks/python/tests/unit/api_objects/dataset/test_rest_operations.py

@@ -0,0 +1,65 @@
+from unittest.mock import Mock
+
+from opik.api_objects.dataset import rest_operations
+
+
+def _find_datasets_returning(*pages) -> Mock:
+    """A rest client whose find_datasets yields the given pages, then an empty one.
+
+    The pagination loop only stops on an empty page, so the trailing empty page
+    is what keeps these tests from spinning forever.
+    """
+    mock_rest_client = Mock()
+    mock_rest_client.datasets.find_datasets.side_effect = [
+        Mock(content=list(page)) for page in (*pages, ())
+    ]
+    return mock_rest_client
+
+
+def _backend_dataset(name: str, type_: str, items_total: int) -> Mock:
+    dataset_fern = Mock()
+    dataset_fern.configure_mock(
+        name=name,
+        description="",
+        type=type_,
+        dataset_items_count=items_total,
+    )
+    return dataset_fern
+
+
+def test_get_test_suites__suite_holds_backend_items__first_insert_syncs_hashes():
+    """A listed suite must not treat its empty local hash set as authoritative.
+
+    Otherwise the first deduplicated insert compares against nothing, decides
+    every item is new, and resubmits items the suite already holds.
+    """
+    mock_rest_client = _find_datasets_returning(
+        [_backend_dataset("my-suite", "evaluation_suite", items_total=25)]
+    )
+
+    suites = rest_operations.get_test_suites(
+        project_name="Test project",
+        rest_client=mock_rest_client,
+    )
+
+    assert len(suites) == 1
+    assert not suites[0]._dataset.__internal_api__hashes_synced__, (
+        "A suite listed from the backend has items we have not hashed locally, "
+        "so the first insert must sync before deduplicating"
+    )
+
+
+def test_get_test_suites__non_suite_datasets__are_skipped():
+    mock_rest_client = _find_datasets_returning(
+        [
+            _backend_dataset("plain-dataset", "dataset", items_total=3),
+            _backend_dataset("my-suite", "evaluation_suite", items_total=3),
+        ]
+    )
+
+    suites = rest_operations.get_test_suites(
+        project_name="Test project",
+        rest_client=mock_rest_client,
+    )
+
+    assert [suite.name for suite in suites] == ["my-suite"]