[NA] [SDK] feat: allow dataset item insertion without deduplication - #8113
Conversation
Add a `deduplication` flag to every Python SDK method that writes dataset items. With `deduplication=False` the whole dedup path is bypassed: the existing items are not downloaded from the backend and no content hashes are computed or compared, which is significantly faster on large datasets. Also raise the default `num_threads` for `Dataset.insert` from 1 to 4, and cache the backend-version probe that gates parallel upload so a loop of inserts does not pay a round trip per call. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
github-actions
Bot
added
documentation
⏱️ pre-commit per-hook timing
⏭️ 36 skipped (no matching files changed)
|
|
🌿 Preview your docs: https://opik-preview-01a0617e-ae89-74ac-9238-e61dfd49d608.docs.buildwithfern.com/docs/opik No broken links found Unverified links (timeout / rate-limited / server error — not failing the check)• https://aistudio.google.com/apikey (401) 📌 Results for commit b2d5c39 |
|
This change looks worth a test.
Would target What it would check
Deploying a test environment for this PR and exploring it — results will follow in a comment. Also already tested. The also touches Python SDK Advisory, from the QA test radar. Nothing here blocks this PR, and anything it proposes is a draft for review. |
|
🔄 Test environment deployment process has started Phase 1: Deploying base version You can monitor the progress here. |
![baz-reviewer[bot]](https://avatars.githubusercontent.com/in/933528?s=60&v=4){kind=small}
| @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. |
There was a problem hiding this comment.
_parallel_insert_supported caches a failed self._rest_client.version() probe as False, so later insert() calls remain single-worker after the backend recovers — should we cache only successful checks and re-probe after failures?
Want Baz to fix this for you? Activate Fixer
Other fix methods
Prompt for AI Agents
Before applying, verify this suggestion against the current code. In
`sdks/python/src/opik/api_objects/dataset/dataset.py` around lines 614-625, replace the
`@functools.cached_property` decorator on `_parallel_insert_supported` with manual
caching that stores the result only after successfully determining backend support
(unsupported/invalid version can be cached as `False`). On a transient exception from
`self._rest_client.version()` (e.g., unreachable backend), return `False` without
caching, so subsequent `insert()` calls can re-probe and recover parallel insert once
the backend becomes available again.
| constants.MIN_BACKEND_VERSION_FOR_PARALLEL_INSERT, | ||
| exc_info=True, | ||
| ) | ||
| return 1 | ||
| return False |
There was a problem hiding this comment.
Transient probe disables parallel uploads
Failures from self._rest_client.version() become False and are cached by @functools.cached_property, so one transient probe failure forces later insert() calls on that Dataset to set num_threads to 1 permanently — should we cache only successful version/support results?
Want Baz to fix this for you? Activate Fixer
Other fix methods
Prompt for AI Agents
Before applying, verify this suggestion against the current code. In
`sdks/python/src/opik/api_objects/dataset/dataset.py` around lines 638-641, fix the
`_parallel_insert_supported` probe so transient failures from
`self._rest_client.version()` are not permanently cached as `False`. Replace the blanket
`cached_property` behavior with explicit caching that stores the support result only
after a successful version lookup and comparison; on exceptions, return `False` for that
insert without populating the cache, allowing later inserts to retry the probe.
| if deduplication: | ||
| items_to_send = self._deduplicate(items) |
There was a problem hiding this comment.
Listed suites bypass backend deduplication
Datasets returned by rest_operations.get_test_suites() retain _hashes_synced=True, so _deduplicate() skips __internal_api__sync_hashes__, treats the empty local hash set as authoritative, and resubmits existing items — should we clear the flag before insertion?
Want Baz to fix this for you? Activate Fixer You can also update your AI coding guidelines based on this comment by apply pr to [branch name]
Other fix methods
Prompt for AI Agents
Before applying, verify this suggestion against the current code. In
sdks/python/src/opik/api_objects/dataset/dataset.py around lines 721-722, the
deduplication path assumes datasets fetched from the backend have their hashes already
synced if `_hashes_synced=True`, but test suite datasets created by
`rest_operations.get_test_suites()` have `_hashes_synced=True` with empty `_hashes`,
causing duplicate items to be submitted to the backend. Ensure that the `_deduplicate()`
method or the datasets from `get_test_suites()` maintain a consistent invariant: either
reset `_hashes_synced=False` for test suite datasets before deduplication, or modify
`_deduplicate()` to always sync hashes for datasets fetched from the backend regardless
of the initial `_hashes_synced` state.
| # 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 |
There was a problem hiding this comment.
Concurrent bypass creates duplicates
Marking _hashes_synced stale before the upload does not synchronize with the bypassed REST batch, so a concurrent deduplicating insert can miss in-flight content and persist a duplicate item ID — should we add per-Dataset synchronization around the state transition and upload, or make snapshots wait for overlapping bypass writes?
Want Baz to fix this for you? Activate Fixer
Other fix methods
Prompt for AI Agents
Before applying, verify this suggestion against the current code. In
`sdks/python/src/opik/api_objects/dataset/dataset.py` around lines 724-727, update
`__internal_api__insert_items_as_dataclasses__` so setting `_hashes_synced = False`
cannot race with a concurrent deduplicating insert. Add a per-Dataset synchronization
mechanism that covers hash syncing/deduplication and the REST batch upload, ensuring
deduplicating inserts wait for any bypassed upload to finish and that the cache is
invalidated consistently afterward. Add or update concurrency tests to verify that
overlapping `deduplication=False` and deduplicating inserts cannot persist the same
content under different IDs.
| def insert( | ||
| self, | ||
| items: Sequence[Dict[str, Any]], | ||
| num_threads: int = 4, | ||
| deduplication: bool = True, | ||
| ) -> None: |
There was a problem hiding this comment.
deduplication accepts non-bool truthy/falsy values, so "false" enables deduplication while 0 or None disables hash synchronization, causing inserts to be duplicated or skipped — should we reject non-bool values with a clear ValueError/TypeError before probing, hashing, or uploading, including in update, read_json, read_jsonl_from_file, and insert_from_pandas?
Want Baz to fix this for you? Activate Fixer
Other fix methods
Prompt for AI Agents
Before applying, verify this suggestion against the current code. In
`sdks/python/src/opik/api_objects/dataset/dataset.py`, update `Dataset.insert` (around
lines 744-749) to validate that `deduplication` is strictly a boolean before probing
backend support, hashing, or uploading, raising a clear `ValueError` or `TypeError` for
values such as strings, integers, or `None` instead of relying on truthiness. Apply the
same runtime validation contract to the forwarding methods `update`, `read_json`,
`read_jsonl_from_file`, and `insert_from_pandas` (around lines 816-834, 933-956,
960-980, and 984-1006), using a shared helper if appropriate, while preserving the
documented default of `True`.
| """ | ||
| 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. |
There was a problem hiding this comment.
Hidden cache resynchronization behavior
The public insert docs describe deduplication=False only as skipping duplicate checks, so callers may miss that it marks the local hash cache out of sync and triggers a backend resync on the next deduplicated insert — should we document this behavior?
Want Baz to fix this for you? Activate Fixer
Other fix methods
Prompt for AI Agents
Before applying, verify this suggestion against the current code. In
`sdks/python/src/opik/api_objects/dataset/dataset.py` around lines 750-759, update the
`Dataset.insert` docstring’s `deduplication` description. Explicitly state that
disabling deduplication marks the local deduplication/hash cache as out of sync, and
that the next insert with deduplication enabled performs a backend hash
resynchronization, adding the resulting network cost and cache behavior for callers.
|
✅ Test environment is now available! To configure additional Environment variables for your environment, run [Deploy Opik AdHoc Environment workflow] (https://github.com/comet-ml/comet-deployment/actions/workflows/deploy_opik_adhoc_env.yaml) Access Information
The deployment has completed successfully and the version has been verified. |
Python SDK E2E Tests Results (Python 3.14)296 tests 287 ✅ 4m 51s ⏱️ For more details on these failures, see this check. Results for commit c684b65. |
…on (#8115) * [NA] [SDK] fix: address review comments on dataset insert deduplication Follow-up to #8113. - `_parallel_insert_supported` cached a probe that failed to reach the backend as `False`, which pinned the dataset to sequential uploads for the rest of the session. Only a conclusive answer is cached now; an unreachable backend is re-probed on the next insert, while a version that cannot be parsed still caches since it will not change. - Test suites listed by `get_test_suites` kept `_hashes_synced=True` with an empty hash set, so the first deduplicated insert compared against nothing and resubmitted items the suite already held. They now start unsynced, matching `get_datasets` and `get_test_suite`. - Reject a non-bool `deduplication`, so a truthy string or `None` cannot silently select the wrong duplicate-checking behaviour. - Note the resync cost of `deduplication=False` in the `insert` docstring. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> * fix(dataset): gate worker count in the insert funnel and test dedup by behaviour Addresses the review comments on the follow-up. - The parallel-upload version gate moves from `insert` into `__internal_api__insert_items_as_dataclasses__`, so a direct caller passing `num_threads > 1` can no longer reach the thread pool against a backend that races on batches sharing a batch_group_id. - The probe-recovery test now spies on `_send_batches` and asserts the worker count that actually reached the upload, instead of only counting probes. - The listed-suite regression test now inserts an item the suite already holds and asserts it is left out of the REST batch, rather than inspecting the private `_hashes_synced` flag. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> * fix(dataset): validate num_threads at the insert funnel and assert real uploads - The funnel compared `num_threads > 1` before checking its type, so a direct caller passing a non-orderable value got a `TypeError` from the gate rather than the `ValueError` naming the argument. It now validates both parameters it consumes. - The sequential-gate test spied on `_send_batches` without calling through, so it only covered argument plumbing. It now calls through and asserts every item still reached the backend exactly once. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> * docs(dataset): document the ValueError raised by insert's argument checks Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> --------- Co-authored-by: Claude Opus 5 (1M context) <noreply@anthropic.com>
2 participants
Details
Adds a
deduplicationflag to every Python SDK method that writes dataset items. Withdeduplication=Falsethe whole dedup path is bypassed — the existing items are not downloaded from the backend, and no content hashes are computed or compared — so inserts into large datasets stop paying for a full item download.Dataset.insert/update/insert_from_json/read_jsonl_from_file/insert_from_pandas, the internal__internal_api__insert_items_as_dataclasses__, and the equivalentTestSuitemethods.deduplication=Trueinsert re-syncs from the backend instead of silently creating real duplicates.Dataset.insert'snum_threadsdefault goes from 1 to 4. The internal dataclass method stays at 1 because it has no backend-version gate, and fanning out parallel batches against older backends races on a sharedbatch_group_id.Dataset— otherwise a loop of inserts would pay one extra round trip per call.Change checklist
Issues
AI-WATERMARK
AI-WATERMARK: yes
num_threadsdefault change were revised on review feedback.Testing
Commands run:
pytest tests/unitinsdks/python— 5125 passed, 2 skipped.pre-commit run --files <changed files>— ruff, ruff-format and mypy pass.Scenarios covered by new unit tests:
deduplication=False, and dropped when it is on.updatewith unchanged content is still sent when dedup is disabled.insert()with nonum_threadsuploads batches concurrently, and the version gate is probed once per dataset across repeated inserts.deduplicationflag is forwarded fromTestSuite.insertto the dataset.The concurrency test was mutation-checked by temporarily restoring the old default of 1 — it fails on the barrier timeout, confirming it is not passing vacuously.
Not run: e2e/library-integration suites, which need a live backend.
Documentation
manage_datasets.mdxgains a "Disabling deduplication" section and a note about the new default thread count.