diff --git a/apps/opik-documentation/documentation/fern/docs-v2/evaluation/advanced/manage_datasets.mdx b/apps/opik-documentation/documentation/fern/docs-v2/evaluation/advanced/manage_datasets.mdx
index 368d0368a67..cf8bce10a4d 100644
--- a/apps/opik-documentation/documentation/fern/docs-v2/evaluation/advanced/manage_datasets.mdx
+++ b/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:
diff --git a/sdks/python/src/opik/api_objects/dataset/dataset.py b/sdks/python/src/opik/api_objects/dataset/dataset.py
index ccff2e9658f..db238ec5906 100644
--- a/sdks/python/src/opik/api_objects/dataset/dataset.py
+++ b/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,11 +684,10 @@ 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
@@ -693,7 +695,6 @@ def __internal_api__insert_items_as_dataclasses__(
         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:
         """
diff --git a/sdks/python/src/opik/api_objects/dataset/test_suite/test_suite.py b/sdks/python/src/opik/api_objects/dataset/test_suite/test_suite.py
index 8894ad999d4..013b38af4e8 100644
--- a/sdks/python/src/opik/api_objects/dataset/test_suite/test_suite.py
+++ b/sdks/python/src/opik/api_objects/dataset/test_suite/test_suite.py
@@ -434,6 +434,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:
         """
         Insert test suite items from a JSON string.
@@ -452,17 +453,23 @@ def insert_from_json(
             keys_mapping: Maps JSON keys to the target keys listed above.
                 Example: ``{"test_data": "data", "checks": "assertions"}``
             ignore_keys: Keys in the JSON dicts to skip during import.
+            deduplication: Whether to skip items whose content already exists in
+                the suite. 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
 
-        self.insert(converters.from_json(json_array, keys_mapping, ignore_keys))
+        self.insert(
+            converters.from_json(json_array, keys_mapping, ignore_keys),
+            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:
         """
         Insert test suite items from a pandas DataFrame.
@@ -483,17 +490,23 @@ def insert_from_pandas(
             keys_mapping: Maps column names to the target keys listed above.
                 Example: ``{"test_data": "data", "checks": "assertions"}``
             ignore_keys: Column names in the DataFrame to skip during import.
+            deduplication: Whether to skip items whose content already exists in
+                the suite. 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
 
-        self.insert(converters.from_pandas(dataframe, keys_mapping, ignore_keys))
+        self.insert(
+            converters.from_pandas(dataframe, keys_mapping, ignore_keys),
+            deduplication=deduplication,
+        )
 
     def insert_from_jsonl_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 items into the test suite.
@@ -512,11 +525,16 @@ def insert_from_jsonl_file(
             keys_mapping: Maps JSON keys to the target keys listed above.
                 Example: ``{"test_data": "data", "checks": "assertions"}``
             ignore_keys: Keys in the JSON objects to skip during import.
+            deduplication: Whether to skip items whose content already exists in
+                the suite. 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
 
-        self.insert(converters.from_jsonl_file(file_path, keys_mapping, ignore_keys))
+        self.insert(
+            converters.from_jsonl_file(file_path, keys_mapping, ignore_keys),
+            deduplication=deduplication,
+        )
 
     def update_test_settings(
         self,
@@ -643,6 +661,7 @@ def get_global_assertions(self) -> List[str]:
     def update(
         self,
         items: List[suite_types.TestSuiteItem],
+        deduplication: bool = True,
     ) -> None:
         """
         Update existing items in the test suite.
@@ -653,6 +672,8 @@ def update(
 
         Args:
             items: List of item dicts to update. Each must contain ``"id"``.
+            deduplication: Whether to skip items whose content already exists in
+                the suite. See :meth:`insert` for details.
 
         Raises:
             DatasetItemUpdateOperationRequiresItemId: If any item is missing
@@ -664,17 +685,22 @@ def update(
                     "Missing id for test suite item to update: %s", item
                 )
 
-        self.insert(items)
+        self.insert(items, deduplication=deduplication)
 
     def insert(
         self,
         items: List[suite_types.TestSuiteItem],
+        deduplication: bool = True,
     ) -> None:
         """
         Insert test cases into the test suite.
 
         Args:
             items: List of test case items to add.
+            deduplication: Whether to skip items whose content already exists
+                in the suite. Pass ``False`` to insert every item as-is
+                without any duplicate checking, which is significantly faster
+                on large suites.
 
         Example:
             >>> suite.insert([
@@ -689,7 +715,9 @@ def insert(
         validators.validate_suite_items(items)
 
         ds_items = [converters.suite_item_dict_to_dataset_item(item) for item in items]
-        self._dataset.__internal_api__insert_items_as_dataclasses__(ds_items)
+        self._dataset.__internal_api__insert_items_as_dataclasses__(
+            ds_items, deduplication=deduplication
+        )
 
     def __internal_api__run_optimization_suite__(
         self,
diff --git a/sdks/python/tests/unit/api_objects/dataset/test_dataset_client.py b/sdks/python/tests/unit/api_objects/dataset/test_dataset_client.py
index 67ce7a3bbcf..a60758cc29c 100644
--- a/sdks/python/tests/unit/api_objects/dataset/test_dataset_client.py
+++ b/sdks/python/tests/unit/api_objects/dataset/test_dataset_client.py
@@ -1,6 +1,7 @@
 import threading
 import time
-from unittest.mock import Mock
+from typing import Optional
+from unittest.mock import Mock, patch
 
 import pytest
 
@@ -112,6 +113,102 @@ def test_insert_deduplication__three_dicts_passed__one_unique__two_duplicates__t
     assert len(inserted_rest_items) == 2, "Two items should be inserted"
 
 
+def test_insert__deduplication_disabled__duplicates_are_inserted():
+    mock_rest_client = Mock()
+
+    dataset = Dataset(
+        name="test_dataset",
+        description="Test description",
+        project_name="Test project",
+        rest_client=mock_rest_client,
+    )
+
+    item_dict = {
+        "input": {"key": "value"},
+        "expected_output": {"key": "output"},
+        "metadata": {"key": "meta"},
+    }
+
+    dataset.insert([item_dict, item_dict], deduplication=False)
+
+    call_args = mock_rest_client.datasets.create_or_update_dataset_items.call_args
+    assert len(call_args[1]["items"]) == 2, (
+        "Both identical items must be sent when deduplication is disabled"
+    )
+
+
+def test_insert__deduplication_disabled__backend_items_are_not_downloaded():
+    mock_rest_client = Mock()
+
+    dataset = Dataset(
+        name="test_dataset",
+        description="Test description",
+        project_name="Test project",
+        rest_client=mock_rest_client,
+    )
+    # The state `get_dataset`/`get_datasets` leave behind: the backend holds
+    # items this object has not hashed yet.
+    dataset.__internal_api__hashes_synced__ = False
+
+    dataset.insert(_make_items(3), deduplication=False)
+
+    mock_rest_client.datasets.stream_dataset_items.assert_not_called()
+
+
+def test_insert__deduplication_disabled__next_deduplicated_insert_syncs_hashes():
+    mock_rest_client = Mock()
+
+    dataset = Dataset(
+        name="test_dataset",
+        description="Test description",
+        project_name="Test project",
+        rest_client=mock_rest_client,
+    )
+
+    dataset.insert(_make_items(3), deduplication=False)
+    assert not dataset.__internal_api__hashes_synced__, (
+        "Skipping deduplication leaves the local hash cache stale"
+    )
+
+    with patch.object(dataset, "__internal_api__sync_hashes__") as sync_hashes:
+        dataset.insert(_make_items(3))
+
+    sync_hashes.assert_called_once()
+
+
+def test_update__deduplication_disabled__unchanged_item_is_still_sent():
+    mock_rest_client = Mock()
+
+    dataset = Dataset(
+        name="test_dataset",
+        description="Test description",
+        project_name="Test project",
+        rest_client=mock_rest_client,
+    )
+
+    item = {
+        "input": {"key": "value"},
+        "expected_output": {"key": "output"},
+        "metadata": {"key": "meta"},
+    }
+    dataset.insert([item])
+
+    inserted_id = mock_rest_client.datasets.create_or_update_dataset_items.call_args[1][
+        "items"
+    ][0].id
+
+    dataset.update([{"id": inserted_id, **item}], deduplication=False)
+
+    assert mock_rest_client.datasets.create_or_update_dataset_items.call_count == 2
+    updated_items = mock_rest_client.datasets.create_or_update_dataset_items.call_args[
+        1
+    ]["items"]
+    assert len(updated_items) == 1, (
+        "An update with unchanged content must still be sent when deduplication "
+        "is disabled"
+    )
+
+
 def test_update__happyflow():
     mock_rest_client = Mock()
 
@@ -312,8 +409,9 @@ def test_insert__invalid_num_threads__raises_before_upload(bad_value):
 def _batches_overlapped(
     monkeypatch,
     mock_rest_client: Mock,
-    num_threads: int,
+    num_threads: Optional[int],
     expect_overlap: bool,
+    item_count: int = _GATE_ITEM_COUNT,
 ) -> bool:
     """Insert through the public API and report whether batches ran concurrently.
 
@@ -321,6 +419,10 @@ def _batches_overlapped(
     boundary rather than at the gate's internal decision — so this still fails
     if insert() stops honouring the worker count downstream.
 
+    ``num_threads=None`` omits the argument so the call exercises the default.
+    ``item_count`` sets how many batches there are, which must not exceed the
+    worker count when overlap is expected — the barrier is sized to it.
+
     ``expect_overlap`` selects how uploads are held, because the two
     expectations fail in opposite directions and need opposite instruments.
 
@@ -339,12 +441,13 @@ def _batches_overlapped(
     hold once per batch.
     """
     _small_batches(monkeypatch, size=_GATE_BATCH_SIZE)
+    batch_count = item_count // _GATE_BATCH_SIZE
 
     lock = threading.Lock()
     in_flight = 0
     peak_in_flight = 0
     barrier = (
-        threading.Barrier(_GATE_BATCH_COUNT, timeout=_OVERLAP_TIMEOUT_SECONDS)
+        threading.Barrier(batch_count, timeout=_OVERLAP_TIMEOUT_SECONDS)
         if expect_overlap
         else None
     )
@@ -375,11 +478,15 @@ def tracked_upload(*args, **kwargs):
         project_name="Test project",
         rest_client=mock_rest_client,
     )
-    dataset.insert(_make_items(_GATE_ITEM_COUNT), num_threads=num_threads)
+    items = _make_items(item_count)
+    if num_threads is None:
+        dataset.insert(items)
+    else:
+        dataset.insert(items, num_threads=num_threads)
 
     assert (
         mock_rest_client.datasets.create_or_update_dataset_items.call_count
-        == _GATE_BATCH_COUNT
+        == batch_count
     ), "Every batch must be uploaded regardless of the thread count"
 
     return peak_in_flight > 1
@@ -451,5 +558,39 @@ def test_insert__sequential__uploads_sequentially_without_probing_version(monkey
         monkeypatch, mock_rest_client, num_threads=1, expect_overlap=False
     ), "num_threads=1 must upload sequentially"
 
-    # The default path must not pay an extra request.
+    # An explicitly sequential upload cannot race, so it must not pay the probe.
     mock_rest_client.version.assert_not_called()
+
+
+# 4 batches, which is the default worker count — so all of them fit in flight
+# together and the barrier can prove the default really uses the pool.
+_DEFAULT_THREADS_ITEM_COUNT = _GATE_BATCH_SIZE * 4
+
+
+def test_insert__num_threads_not_given__uploads_concurrently_by_default(monkeypatch):
+    assert _batches_overlapped(
+        monkeypatch,
+        _mock_rest_client(),
+        num_threads=None,
+        expect_overlap=True,
+        item_count=_DEFAULT_THREADS_ITEM_COUNT,
+    ), "insert() must upload in parallel without being asked to"
+
+
+def test_insert__repeated_inserts__backend_version_probed_once(monkeypatch):
+    _small_batches(monkeypatch, size=_GATE_BATCH_SIZE)
+    mock_rest_client = _mock_rest_client()
+    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, (
+        "Parallel upload is the default, so the version gate must be probed once "
+        "per dataset rather than once per insert"
+    )
diff --git a/sdks/python/tests/unit/api_objects/dataset/test_suite/test_test_suite.py b/sdks/python/tests/unit/api_objects/dataset/test_suite/test_test_suite.py
index 310cc0fc299..09531e2d997 100644
--- a/sdks/python/tests/unit/api_objects/dataset/test_suite/test_test_suite.py
+++ b/sdks/python/tests/unit/api_objects/dataset/test_suite/test_test_suite.py
@@ -147,6 +147,22 @@ def test_insert__with_assertions_shorthand__creates_evaluator_items(self):
         assert len(item.evaluators) == 1
         assert item.evaluators[0].type == "llm_judge"
 
+    @pytest.mark.parametrize("deduplication", [True, False])
+    def test_insert__deduplication_flag__forwarded_to_the_dataset(self, deduplication):
+        mock_dataset = _create_mock_dataset()
+        suite = test_suite.TestSuite(
+            name="test_suite",
+            dataset_=mock_dataset,
+        )
+
+        suite.insert(
+            [{"data": {"input": "test"}}],
+            deduplication=deduplication,
+        )
+
+        call_args = mock_dataset.__internal_api__insert_items_as_dataclasses__.call_args
+        assert call_args[1]["deduplication"] is deduplication
+
     def test_resolve_evaluators__with_both_assertions_and_evaluators__raises_value_error(
         self,
     ):