[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,25 @@ 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, and
+        # a non-integer worker count would fail on the comparison below with a
+        # TypeError instead of naming the offending argument.
+        if not isinstance(deduplication, bool):
+            raise ValueError("deduplication must be a bool")
+        if isinstance(num_threads, bool) or not isinstance(num_threads, int):
+            raise ValueError("num_threads must be a positive integer")
+        if num_threads < 1:
+            raise ValueError("num_threads must be a positive integer")
+
+        # 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,22 +796,27 @@ 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
                 dataset version. If a batch fails the call raises, and the
                 batches that already succeeded stay persisted. Older Opik
                 backends do not support parallel upload and fall back to a
                 sequential one.
+
+        Raises:
+            ValueError: If ``num_threads`` is not a positive integer, or
+                ``deduplication`` is not a bool.
         """
         if isinstance(num_threads, bool) or not isinstance(num_threads, int):
             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)

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

@@ -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,147 @@ 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,
+    )
+
+    # The spy calls through, so the upload still happens and the assertions
+    # below cover the items actually reaching the backend, not just the
+    # argument the gate computed.
+    used_workers = []
+    original_send = Dataset._send_batches
+
+    def spy_send_batches(self, batches, batch_group_id, num_threads):
+        used_workers.append(num_threads)
+        return original_send(self, batches, batch_group_id, num_threads)
+
+    monkeypatch.setattr(Dataset, "_send_batches", spy_send_batches)
+
+    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"
+    )
+
+    create_or_update = mock_rest_client.datasets.create_or_update_dataset_items
+    submitted = sorted(
+        item.data["input"]["i"]
+        for call in create_or_update.call_args_list
+        for item in call.kwargs["items"]
+    )
+    assert submitted == [0, 1, 2, 3], (
+        "Forcing a sequential upload must still deliver every item exactly once"
+    )
+
+
+@pytest.mark.parametrize("bad_value", [0, -1, 1.5, "2", True, None])
+def test_internal_insert__invalid_num_threads__raises_value_error(bad_value):
+    """Direct callers get the named ValueError, not a TypeError from the gate."""
+    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="num_threads must be a positive integer"):
+        dataset.__internal_api__insert_items_as_dataclasses__(
+            [dataset_item.DatasetItem(**item) for item in _make_items(2)],
+            num_threads=bad_value,
+        )
+
+    mock_rest_client.datasets.create_or_update_dataset_items.assert_not_called()
+
+
+@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()