[NA] [SDK] fix: address review comments on dataset insert deduplication - #8115
Conversation
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>
github-actions
Bot
added
python
⏱️ pre-commit per-hook timing
⏭️ 39 skipped (no matching files changed)
|
|
Already covered by a test in this PR. The real user-facing fix here is the one line in also touches Python SDK Advisory, from the QA test radar. Nothing here blocks this PR, and anything it proposes is a draft for review. Re-checked after a push on 02 Sep 12:06 UTC — nothing the verdict depends on changed. |
|
🔄 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}
| if not isinstance(deduplication, bool): | ||
| raise ValueError("deduplication must be a bool") |
There was a problem hiding this comment.
Internal inserts bypass backend safety gate
The internal insertion funnel validates deduplication but forwards num_threads to _send_batches without applying the _parallel_insert_supported gate, so direct calls with num_threads > 1 use ThreadPoolExecutor against old backends instead of the fallback in insert() — should we apply the same worker-count compatibility check here?
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 745-746, update
`__internal_api__insert_items_as_dataclasses__` so it applies the same
`_parallel_insert_supported` compatibility gate as `insert()`. Before forwarding
`num_threads` to `_send_batches`, reduce values greater than 1 to 1 when the backend
does not support parallel insertion, ensuring direct internal callers cannot bypass the
sequential-upload fallback.
There was a problem hiding this comment.
Commit 00c79c8 addressed this comment by applying the _parallel_insert_supported gate inside the internal insertion funnel before forwarding num_threads to batch sending. Unsupported backends now force worker count to 1, including for direct internal callers.
| 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 dataset._parallel_insert_supported, ( | ||
| "Once the backend answers, parallel upload must be available again" | ||
| ) |
There was a problem hiding this comment.
Parallel upload regression goes undetected
test_insert__version_probe_recovers__parallel_upload_resumes only checks the probe count and _parallel_insert_supported, so a sequential second insert would still pass — should we add a barrier or active-worker counter to the batch upload mock to assert overlap while retaining the probe-retry assertions?
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/tests/unit/api_objects/dataset/test_dataset_client.py` around lines
569-577, strengthen `test_insert__version_probe_recovers__parallel_upload_resumes` so it
verifies that the second insert actually uses parallel/concurrent batch uploading, not
merely that the `_parallel_insert_supported` cache flag is set and the probe call count
is correct. Instrument the batch upload mock with a barrier, event, or deterministic
active-concurrent-call counter, and assert that overlap occurs among batch uploads
during the second insert, while retaining the existing version-probe retry assertions.
Alternatively, if adding this determinism proves too complex, consider renaming the test
to reflect that it only covers probe-cache recovery.
There was a problem hiding this comment.
Addressed in 00c79c8. test_insert__version_probe_recovers__parallel_upload_resumes now installs a spy on _send_batches that calls through to the real implementation and records the worker count each insert actually used, then asserts [0] == 1 (sequential while the backend is unreachable) and [1] > 1 (workers again once it answers). A second insert that stayed sequential now fails the test, which was the gap you flagged.
There was a problem hiding this comment.
Addressed. The spy now verifies the effective worker count reaching _send_batches: the first insert uses 1, while the recovered second insert uses multiple workers, and _send_batches fans those workers out through ThreadPoolExecutor.
| # 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 |
There was a problem hiding this comment.
Listed suites lose owning project scope
In get_test_suites, the lazy-sync suite_dataset keeps the caller's project_name even when dataset_fern resolves via project_id, so workspace-wide or fallback listings can stream or upsert same-named suites into the wrong dataset — should we resolve and retain the returned suite's owning project name before enabling __internal_api__hashes_synced__, as Dataset.from_public already does?
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/rest_operations.py` around lines 287-290 in
`get_test_suites`, fix the lazy-sync setup so each listed suite retains its actual
owning project rather than the caller’s `project_name`. Resolve the project name from
`dataset_fern.project_id` (as `Dataset.from_public` does) and pass that resolved name
when constructing `suite_dataset` before enabling `__internal_api__hashes_synced__`,
ensuring streaming and upserts target the correct dataset.
There was a problem hiding this comment.
Not taking this one here, because it is pre-existing rather than introduced by this change.
get_datasets uses the identical pattern — it passes the caller's project_name straight into the Dataset while deriving project_id from that same name, and it sets __internal_api__hashes_synced__ = False too. So the scope behaviour of a listed suite is now exactly the scope behaviour of a listed dataset; before this PR the suite was inconsistent with it in the other direction (it never synced at all, which is the resubmission bug this PR fixes).
Resolving the owning project per listed item, the way from_public does, means a get_project_by_id call for every dataset in the listing. That is precisely the N+1 the lazy-sync design was built to avoid, so it is a deliberate trade-off to change, not a local fix — and it would need to change get_datasets at the same time to keep the two consistent. Worth doing as its own change if we want listings to be project-accurate.
There was a problem hiding this comment.
Thanks for the context—this is an intentional lazy-sync trade-off shared with get_datasets, rather than a regression in this PR. I’ll save this to memory once the PR is merged.
| # 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 |
There was a problem hiding this comment.
Missing listed-suite deduplication coverage
The regression test only checks the private __internal_api__hashes_synced__ flag, so it doesn't verify that a duplicate item is actually omitted from the REST batch after retrieving a populated suite via get_test_suites() — should we extend it, or add an integration/E2E test, to assert the batch excludes the duplicate or the backend item count stays unchanged?
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/rest_operations.py` around lines 287-290,
extend the tests for `get_test_suites()` to cover the cross-file synchronization
behavior introduced by `__internal_api__hashes_synced__ = False`. Populate a suite,
retrieve it through `get_test_suites()`, insert an identical item with deduplication
enabled, and assert that the REST batch omits the duplicate or that the backend item
count remains unchanged; do not limit the test to checking the private flag.
There was a problem hiding this comment.
Commit 00c79c8 addressed this comment by adding an integration-style regression test that retrieves a populated suite, inserts a duplicate and a new item, and asserts only the new item is submitted.
| suites = rest_operations.get_test_suites( | ||
| project_name="Test project", | ||
| rest_client=mock_rest_client, | ||
| ) | ||
|
|
||
| assert len(suites) == 1 | ||
| assert not suites[0]._dataset.__internal_api__hashes_synced__, ( | ||
| "A suite listed from the backend has items we have not hashed locally, " | ||
| "so the first insert must sync before deduplicating" | ||
| ) |
There was a problem hiding this comment.
Suite deduplication behavior remains untested
This test only inspects private _hashes_synced and never performs the first deduplicating insert, so it doesn't verify that listed suites re-read backend items and avoid duplicate submissions — should we run that insert against a mocked backend and assert the item-fetch/sync request and submitted items?
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/tests/unit/api_objects/dataset/test_rest_operations.py` around lines 40-49,
update `test_get_test_suites__suite_holds_backend_items__first_insert_syncs_hashes` so
it performs the suite’s first deduplicating insert instead of only inspecting the
private `_hashes_synced` flag. Configure the mocked backend item-fetch and insert
endpoints, execute an insert containing existing and new items, and assert that backend
items are fetched/synchronized and only the new items are submitted.
There was a problem hiding this comment.
Commit 00c79c8 addressed this comment by exercising the first deduplicating insert with existing and new items, then asserting only the new item is submitted. It does not explicitly assert the fetch call itself.
|
✅ 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. |
…y 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>
| if num_threads > 1 and not self._parallel_insert_supported: | ||
| num_threads = 1 |
There was a problem hiding this comment.
Direct callers get incidental type errors
The internal funnel compares num_threads > 1 before validating its type, so direct callers of __internal_api__insert_items_as_dataclasses__ with non-orderable values get TypeError instead of the documented ValueError — should we move the same positive-integer validation to this boundary or normalize before comparing?
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 752-753, update
`__internal_api__insert_items_as_dataclasses__` so `num_threads` is validated before the
`> 1` comparison; currently `None` or non-orderable values raise an incidental
`TypeError` instead of the documented `ValueError`. Apply the same positive-integer and
boolean-exclusion checks used by `insert`, preferably through a shared private
validator, while preserving the sequential fallback for unsupported parallel inserts.
There was a problem hiding this comment.
Commit 9c86938 addressed this comment by validating num_threads before the > 1 comparison, rejecting booleans, non-integers, and values below 1 with ValueError.
| monkeypatch.setattr( | ||
| Dataset, | ||
| "_send_batches", | ||
| lambda self, batches, batch_group_id, num_threads: used_workers.append( | ||
| num_threads | ||
| ), | ||
| ) |
There was a problem hiding this comment.
Sequential gate test misses upload failures
_send_batches only records num_threads, so __internal_api__insert_items_as_dataclasses__ can pass without submitting or processing batches and the test checks only argument plumbing — could we use a result-preserving fake or mocked backend, assert successful insertion, and make overlapping execution observable to cover the sequential gate?
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/tests/unit/api_objects/dataset/test_dataset_client.py` around lines
627-633, strengthen `test_internal_insert__old_backend__worker_count_still_gated`: the
no-op `_send_batches` stub lets insertion pass without submitting or processing any
batches. Replace it with a result-preserving fake or invoke the real upload path against
a mocked backend, assert that all items are successfully processed, and record batch
execution with synchronization or timing so concurrent execution would be observable.
Keep the assertion that the old backend forces `num_threads` to 1.
There was a problem hiding this comment.
Addressed in 9c86938. The spy now calls through to the real _send_batches instead of swallowing the call, and the test additionally asserts that all four items reached create_or_update_dataset_items exactly once — so it no longer passes if the gate short-circuits the upload. I also added test_internal_insert__invalid_num_threads__raises_value_error covering the rejected worker counts at the same boundary.
I did not go as far as making overlap observable here: that is what test_insert__backend_supports_parallel__batches_uploaded_concurrently and test_insert__backend_older_than_minimum__uploads_sequentially already do with the barrier/hold instrumentation, and duplicating it in this test would assert the same thing twice.
…al 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>
…ecks Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
2 participants
Details
Follow-up to #8113, addressing the Baz review comments left on it. Three behavioural fixes plus a docstring note; one comment declined with reasoning below.
_parallel_insert_supportedwas acached_property, so a probe that could not reach the backend cachedFalseand pinned thatDatasetto sequential uploads for the rest of the session. This was a regression from [NA] [SDK] feat: allow dataset item insertion without deduplication #8113 — before it, the probe ran per insert. Now only a conclusive answer is cached: an unreachable backend returnsFalsewithout caching and is re-probed on the next insert, while a version string the SDK cannot parse still caches, since it will not change.rest_operations.get_test_suitesconstructed itsDatasetdirectly and left_hashes_synced=Truewith an empty hash set, so the first deduplicated insert compared against nothing, judged every item new, and resubmitted items the suite already held. It now starts unsynced, matchingget_datasetsandget_test_suite. This one predates [NA] [SDK] feat: allow dataset item insertion without deduplication #8113 — the lazy-sync design it depends on landed earlier — but it lives in the deduplication path that PR touched.deduplication.deduplication="false"silently enabled deduplication andNone/0silently disabled it. Rejected with aValueErrornow, validated both ininsert(before the version probe, so bad input costs no request) and in__internal_api__insert_items_as_dataclasses__, which is the funnel every path includingTestSuite.insertgoes through.deduplication=Falsemakes the next deduplicating insert re-read the dataset's items.Declined: per-Dataset locking around the bypass write
Baz asked for synchronization so a
deduplication=Falseinsert cannot race a concurrent deduplicating insert on the sameDataset. Not taken:Datasethas never been thread-safe —_hashes,_id_to_hashand_dataset_items_countare all unsynchronized today — so this would be a new concurrency contract rather than a fix, and a lock spanning the REST upload would serialize the parallel-batch upload the same class just gained. The existing ordering is also the safe one:_hashes_syncedis cleared before the upload, so a failed or partial upload leaves the cache invalidated rather than falsely marked fresh. Worth a separate discussion if we want to declareDatasetthread-safe.Change checklist
Issues
AI-WATERMARK
AI-WATERMARK: yes
Testing
Commands run:
pytest tests/unitinsdks/python— 5135 passed, 2 skipped.pre-commit run --files <changed files>— ruff, ruff-format and mypy pass.New unit tests:
deduplicationvalues ("false",0,1,None,"",[]) raise before any request is made.get_test_suitesreturns suites whose first insert will sync hashes, and still skips non-suite datasets.The
get_test_suitestest was mutation-checked by removing the fix — it fails, confirming it covers the reported defect.Documentation
No documentation changes. The public behaviour documented in
manage_datasets.mdxby #8113 is unchanged; this PR only adds a one-sentence note to theDataset.insertdocstring about the resync cost ofdeduplication=False.