comet-ml · alexkuzmik · Sep 2, 2026 · Sep 2, 2026 · Sep 2, 2026 · Sep 2, 2026
@@ -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,19 @@ 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")
+
+        # Gated here rather than in `insert` so every caller of this funnel is
+        # covered: older backends race on concurrent batches that share a
+        # batch_group_id, and a direct caller asking for workers must not be
+        # able to skip that check.
+        if num_threads > 1 and not self._parallel_insert_supported:
+            num_threads = 1
+
         if deduplication:
             items_to_send = self._deduplicate(items)
         else:
@@ -756,7 +790,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,9 +804,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")
-
-        if num_threads > 1 and not self._parallel_insert_supported:
-            num_threads = 1
+        # Checked here too so bad input raises before any item is converted.
+        if not isinstance(deduplication, bool):
+            raise ValueError("deduplication must be a bool")
 
         dataset_items: List[dataset_item.DatasetItem] = [  # type: ignore
             (dataset_item.DatasetItem(**item) if isinstance(item, dict) else item)

@@ -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(

@@ -6,6 +6,7 @@
 import pytest
 
 from opik.api_objects import constants
+from opik.api_objects.dataset import dataset_item
 from opik.api_objects.dataset.dataset import Dataset
 
 
@@ -551,6 +552,114 @@ 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,
+    )
+
+    # Spying on the upload layer keeps the assertion on the worker count that
+    # actually reached it, so a gate that stops honouring the probe still fails
+    # this test.
+    workers_per_insert = []
+    original_send = Dataset._send_batches
+
+    def spy_send_batches(self, batches, batch_group_id, num_threads):
+        workers_per_insert.append(num_threads)
+        return original_send(self, batches, batch_group_id, num_threads)
+
+    monkeypatch.setattr(Dataset, "_send_batches", spy_send_batches)
+
+    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 workers_per_insert[0] == 1, (
+        "While the backend is unreachable the upload must stay sequential"
+    )
+    assert workers_per_insert[1] > 1, (
+        "Once the backend answers, the upload must go back to using workers"
+    )
+
+
+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"
+    )
+
+
+def test_internal_insert__old_backend__worker_count_still_gated(monkeypatch):
+    """The gate lives in the funnel, so a direct caller cannot skip it."""
+    _small_batches(monkeypatch, size=_GATE_BATCH_SIZE)
+    mock_rest_client = _mock_rest_client("2.2.7")
+    dataset = Dataset(
+        name="test_dataset",
+        description="Test description",
+        project_name="Test project",
+        rest_client=mock_rest_client,
+    )
+
+    used_workers = []
+    monkeypatch.setattr(
+        Dataset,
+        "_send_batches",
+        lambda self, batches, batch_group_id, num_threads: used_workers.append(
+            num_threads
+        ),
+    )
+
+    dataset.__internal_api__insert_items_as_dataclasses__(
+        [dataset_item.DatasetItem(**item) for item in _make_items(4)],
+        num_threads=4,
+    )
+
+    assert used_workers == [1], (
+        "A backend that predates parallel insert must force a sequential upload "
+        "even when the internal API is called directly"
+    )
+
+
+@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()
 

@@ -0,0 +1,88 @@
+from unittest.mock import Mock, patch
+
+from opik.api_objects.dataset import rest_operations
+from opik.rest_api.types import dataset_item as rest_dataset_item
+
+
+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__insert_duplicates_existing_item__duplicate_not_submitted():
+    """A listed suite must re-read its backend items before deduplicating.
+
+    Otherwise the first insert compares against an empty local hash set,
+    decides every item is new, and resubmits items the suite already holds.
+    """
+    existing_content = {"question": "already in the suite"}
+    mock_rest_client = _find_datasets_returning(
+        [_backend_dataset("my-suite", "evaluation_suite", items_total=1)]
+    )
+
+    suites = rest_operations.get_test_suites(
+        project_name="Test project",
+        rest_client=mock_rest_client,
+    )
+    assert len(suites) == 1
+
+    backend_item = rest_dataset_item.DatasetItem(
+        id="existing-item-id", source="sdk", data=existing_content
+    )
+    with patch(
+        "opik.api_objects.dataset.rest_operations.rest_stream_parser.read_and_parse_stream",
+        side_effect=[[backend_item], []],
+    ):
+        suites[0].insert(
+            [
+                {"data": existing_content},
+                {"data": {"question": "brand new"}},
+            ]
+        )
+
+    create_or_update = mock_rest_client.datasets.create_or_update_dataset_items
+    submitted = [
+        item
+        for call in create_or_update.call_args_list
+        for item in call.kwargs["items"]
+    ]
+
+    assert [item.data for item in submitted] == [{"question": "brand new"}], (
+        "The item the suite already holds must be recognised as a duplicate and "
+        "left out of the batch"
+    )
+
+
+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"]