[NA] [SDK] feat: allow dataset item insertion without deduplication

apps/opik-documentation/documentation/fern/docs-v2/evaluation/advanced/manage_datasets.mdx

@@ -277,12 +277,32 @@ dataset.insert([
 <Tip>
   Opik automatically deduplicates items that are inserted into a dataset when using the Python SDK. This means that you
   can insert the same item multiple times without duplicating it in the dataset. This combined with the `get or create
-  dataset` methods means that you can use the SDK to manage your datasets in a "fire and forget" manner.
+  dataset` methods means that you can use the SDK to manage your datasets in a "fire and forget" manner. It can be
+  turned off with `deduplication=False`, see [Disabling deduplication](#disabling-deduplication).
 </Tip>
 
+#### Disabling deduplication
+
+Deduplication requires the Python SDK to download the dataset's existing items once so it can compare their
+content hashes against the items you are inserting. On large datasets that download dominates the insert. If you
+already know your items are unique — for example when populating a fresh dataset, or when you generate ids
+yourself — pass `deduplication=False` to skip that work entirely: nothing is downloaded, no hashes are computed,
+and every item you pass is sent as-is.
+
+```python title="Python" language="python"
+dataset.insert(items, deduplication=False)
+```
+
+The flag is available on every Python SDK method that writes items — `insert`, `update`, `insert_from_json`,
+`insert_from_pandas` and `read_jsonl_from_file` — as well as on the equivalent `TestSuite` methods. With
+deduplication disabled, inserting the same content twice produces two separate dataset items.
+
 <Note>
   When using the SDK to insert items, a new dataset version is automatically created.
   If you insert items in multiple batches within a single `insert()` call, they are grouped into one version.
+  The Python SDK uploads those batches on 4 worker threads by default; use `num_threads=1` to upload them
+  sequentially instead. Parallel upload requires a recent Opik backend — against older ones the SDK falls back
+  to a sequential upload and logs a warning.
 </Note>
 
 Once the items have been inserted, you can view them in the Opik UI:

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

@@ -611,14 +611,18 @@ def _insert_batch_with_retry(
         )
         LOGGER.debug("Successfully sent dataset items batch of size %d", len(batch))
 
-    def _resolve_num_threads(self, num_threads: int) -> int:
-        """Downgrade to a sequential upload when the backend predates parallel support.
+    @functools.cached_property
+    def _parallel_insert_supported(self) -> bool:
+        """Whether the backend tolerates concurrent batches sharing a batch_group_id.
 
-        Older backends race on concurrent batches that share a batch_group_id,
-        so parallelism is only safe from
+        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 fall back to sequential rather than risk the race.
+        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.
         """
         try:
             backend_version = self._rest_client.version()["version"]
@@ -634,7 +638,7 @@ def _resolve_num_threads(self, num_threads: int) -> int:
                 constants.MIN_BACKEND_VERSION_FOR_PARALLEL_INSERT,
                 exc_info=True,
             )
-            return 1
+            return False
 
         if not supported:
             LOGGER.warning(
@@ -644,9 +648,8 @@ def _resolve_num_threads(self, num_threads: int) -> int:
                 backend_version,
                 constants.MIN_BACKEND_VERSION_FOR_PARALLEL_INSERT,
             )
-            return 1
 
-        return num_threads
+        return supported
 
     def _send_batches(
         self,
@@ -681,19 +684,17 @@ def _send_batches(
             for future in futures.as_completed(submitted):
                 future.result()
 
-    def __internal_api__insert_items_as_dataclasses__(
-        self,
-        items: List[dataset_item.DatasetItem],
-        num_threads: int = 1,
-    ) -> None:
+    def _deduplicate(
+        self, items: List[dataset_item.DatasetItem]
+    ) -> List[dataset_item.DatasetItem]:
+        """Drop items whose content hash was already seen locally or on the backend."""
         # Lazy-sync against the backend the first time we insert into a
         # dataset that was fetched from the backend (list or get-by-name
         # factory), so content-hash dedup still works without paying an
         # N+1 sync at list time.
         if not self._hashes_synced:
             self.__internal_api__sync_hashes__()
 
-        # Remove duplicates if they already exist
         deduplicated_items: List[dataset_item.DatasetItem] = []
         for item in items:
             item_hash = item.content_hash()
@@ -709,7 +710,23 @@ def __internal_api__insert_items_as_dataclasses__(
             self._hashes.add(item_hash)
             self._id_to_hash[item.id] = item_hash
 
-        rest_items = [self._convert_to_rest_item(item) for item in deduplicated_items]
+        return deduplicated_items
+
+    def __internal_api__insert_items_as_dataclasses__(
+        self,
+        items: List[dataset_item.DatasetItem],
+        num_threads: int = 1,
+        deduplication: bool = True,
+    ) -> None:
+        if deduplication:
+            items_to_send = self._deduplicate(items)
+        else:
+            # Nothing was hashed, so the local cache no longer describes the
+            # backend; force a re-sync before the next deduplicated insert.
+            items_to_send = items
+            self._hashes_synced = False
+
+        rest_items = [self._convert_to_rest_item(item) for item in items_to_send]
 
         batches = sequence_splitter.split_into_batches(
             rest_items,
@@ -724,42 +741,44 @@ def __internal_api__insert_items_as_dataclasses__(
         # Invalidate the cached count so it will be fetched from backend on next access
         self._dataset_items_count = None
 
-    def insert(self, items: Sequence[Dict[str, Any]], num_threads: int = 1) -> None:
+    def insert(
+        self,
+        items: Sequence[Dict[str, Any]],
+        num_threads: int = 4,
+        deduplication: bool = True,
+    ) -> None:
         """
         Insert new items into the dataset. A new dataset version will be created.
 
         Args:
             items: List of dicts (which will be converted to dataset items)
                 to add to the dataset.
+            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.
             num_threads: Number of worker threads used to upload the item
-                batches. Must be a positive integer. With ``1`` (the default)
-                batches are uploaded sequentially. With more than ``1`` the
-                batches of this single ``insert`` are uploaded in parallel;
-                they all land in one dataset version. If any batch fails the
-                call re-raises; there is no rollback, so batches that already
-                succeeded remain persisted. Items are keyed by their ``id``, so
-                parallel and sequential inserts of the same items produce
-                identical dataset content; the input order is not a read-back
-                guarantee. Requires an Opik backend of at least
-                ``constants.MIN_BACKEND_VERSION_FOR_PARALLEL_INSERT``; against
-                older backends, or when the backend version cannot be
-                determined, the upload falls back to sequential and logs a
-                warning.
+                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.
         """
         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:
-            num_threads = self._resolve_num_threads(num_threads)
+        if num_threads > 1 and not self._parallel_insert_supported:
+            num_threads = 1
 
         dataset_items: List[dataset_item.DatasetItem] = [  # type: ignore
             (dataset_item.DatasetItem(**item) if isinstance(item, dict) else item)
             for item in items
         ]
         self.__internal_api__insert_items_as_dataclasses__(
-            dataset_items, num_threads=num_threads
+            dataset_items, num_threads=num_threads, deduplication=deduplication
         )
 
     @property
@@ -794,12 +813,14 @@ def __internal_api__sync_hashes__(self) -> None:
         self._hashes_synced = True
         LOGGER.debug("Finish hash sync in dataset")
 
-    def update(self, items: List[Dict[str, Any]]) -> None:
+    def update(self, items: List[Dict[str, Any]], deduplication: bool = True) -> None:
         """
         Update existing items in the dataset.
 
         Args:
             items: List of DatasetItem objects to update in the dataset. You need to provide the full item object as it will override what has been supplied previously.
+            deduplication: Whether to skip items whose content already exists in
+                the dataset. See :meth:`insert` for details.
 
         Raises:
             DatasetItemUpdateOperationRequiresItemId: If any item in the list is missing an id.
@@ -810,7 +831,7 @@ def update(self, items: List[Dict[str, Any]]) -> None:
                     "Missing id for dataset item to update: %s", item
                 )
 
-        self.insert(items)
+        self.insert(items, deduplication=deduplication)
 
     def _delete_batch_with_retry(
         self,
@@ -912,6 +933,7 @@ def insert_from_json(
         json_array: str,
         keys_mapping: Optional[Dict[str, str]] = None,
         ignore_keys: Optional[List[str]] = None,
+        deduplication: bool = True,
     ) -> None:
         """
         Args:
@@ -921,6 +943,8 @@ def insert_from_json(
                 Example: {'Expected output': 'expected_output'}
             ignore_keys: if your json dicts contain keys that are not needed for DatasetItem
                 construction - pass them as ignore_keys argument
+            deduplication: Whether to skip items whose content already exists in
+                the dataset. See :meth:`insert` for details.
         """
         keys_mapping = {} if keys_mapping is None else keys_mapping
         ignore_keys = [] if ignore_keys is None else ignore_keys
@@ -929,13 +953,14 @@ def insert_from_json(
             json_array, keys_mapping=keys_mapping, ignore_keys=ignore_keys
         )
 
-        self.insert(new_items)
+        self.insert(new_items, deduplication=deduplication)
 
     def read_jsonl_from_file(
         self,
         file_path: str,
         keys_mapping: Optional[Dict[str, str]] = None,
         ignore_keys: Optional[List[str]] = None,
+        deduplication: bool = True,
     ) -> None:
         """
         Read JSONL from a file and insert it into the dataset.
@@ -946,17 +971,20 @@ def read_jsonl_from_file(
                 Example: {'Expected output': 'expected_output'}
             ignore_keys: if your json dicts contain keys that are not needed for DatasetItem
                 construction - pass them as ignore_keys argument
+            deduplication: Whether to skip items whose content already exists in
+                the dataset. See :meth:`insert` for details.
         """
         keys_mapping = {} if keys_mapping is None else keys_mapping
         ignore_keys = [] if ignore_keys is None else ignore_keys
         new_items = converters.from_jsonl_file(file_path, keys_mapping, ignore_keys)
-        self.insert(new_items)
+        self.insert(new_items, deduplication=deduplication)
 
     def insert_from_pandas(
         self,
         dataframe: "pd.DataFrame",
         keys_mapping: Optional[Dict[str, str]] = None,
         ignore_keys: Optional[List[str]] = None,
+        deduplication: bool = True,
     ) -> None:
         """
         Requires: `pandas` library to be installed.
@@ -967,13 +995,15 @@ def insert_from_pandas(
                 Example: {'Expected output': 'expected_output'}
             ignore_keys: if your dataframe contains columns that are not needed for DatasetItem
                 construction - pass them as ignore_keys argument
+            deduplication: Whether to skip items whose content already exists in
+                the dataset. See :meth:`insert` for details.
         """
         keys_mapping = {} if keys_mapping is None else keys_mapping
         ignore_keys = [] if ignore_keys is None else ignore_keys
 
         new_items = converters.from_pandas(dataframe, keys_mapping, ignore_keys)
 
-        self.insert(new_items)
+        self.insert(new_items, deduplication=deduplication)
 
     def get_version_view(self, version_name: str) -> DatasetVersion:
         """