From 195912d76e7bce1e44deff43e62bfba35be88986 Mon Sep 17 00:00:00 2001 From: t Date: Mon, 24 Aug 2026 10:42:02 +0300 Subject: [PATCH 1/5] [OPIK-7891] [BE] fix: count distinct item ids when sizing a dataset version insertItems returned items.size() -- the raw list length, deliberately not a DB row count, because ClickHouse async inserts report 0 rows before commit. Reads collapse a repeated dataset_item_id to one row via LIMIT 1 BY, so a stable id appearing twice in one batch made the version total one higher than the rows the version actually holds. Every version total flows through this value: createFirstVersion feeds it straight to itemsTotal, and applyDelta sums added + edited + copied. So the fix belongs at the source -- count distinct stable ids, matching what the storage engine keeps. This corrects the mechanism recorded on the ticket. The reported cause was asynchronous ClickHouse visibility racing concurrent batches on the append path; it is neither a race nor on the append path. One single-threaded request reproduces it, and the appending path already classifies a cross-batch duplicate correctly via countExistingItemIds -- verified by a test that passes with and without this change. num_threads was a red herring: the SDK's content-hash dedup drops same-content duplicates before batching, so a duplicate id only survives into a request when the content differs. Tests cover both shapes: a duplicate inside the version-creating batch (fails without this fix) and a duplicate spanning two batches in one batch_group_id (passes either way, pinning the append path's behaviour). Co-Authored-By: Claude Opus 5 (1M context) --- .../opik/domain/DatasetItemVersionDAO.java | 10 ++- .../v1/priv/DatasetVersionResourceTest.java | 90 +++++++++++++++++++ 2 files changed, 99 insertions(+), 1 deletion(-) diff --git a/apps/opik-backend/src/main/java/com/comet/opik/domain/DatasetItemVersionDAO.java b/apps/opik-backend/src/main/java/com/comet/opik/domain/DatasetItemVersionDAO.java index a7240e24355..cedf2d7fd85 100644 --- a/apps/opik-backend/src/main/java/com/comet/opik/domain/DatasetItemVersionDAO.java +++ b/apps/opik-backend/src/main/java/com/comet/opik/domain/DatasetItemVersionDAO.java @@ -3683,7 +3683,15 @@ public Mono insertItems(@NonNull UUID datasetId, @NonNull UUID newVersionI // Note: ClickHouse with async inserts returns 0 immediately before commit. // We return the count of items we're inserting instead of relying on getRowsUpdated. - long itemCount = items.size(); + // + // Count DISTINCT stable ids, not rows handed in (OPIK-7891): reads collapse a repeated + // dataset_item_id to one row via LIMIT 1 BY, so counting the raw list inflates every + // version total derived from this value. Callers set datasetItemId before inserting; + // fall back to id so an unnormalized item still counts as itself rather than as null. + long itemCount = items.stream() + .map(item -> item.datasetItemId() != null ? item.datasetItemId() : item.id()) + .distinct() + .count(); return asyncTemplate.nonTransaction(connection -> { Segment segment = startSegment(DATASET_ITEM_VERSIONS, CLICKHOUSE, "insert_delta_items"); diff --git a/apps/opik-backend/src/test/java/com/comet/opik/api/resources/v1/priv/DatasetVersionResourceTest.java b/apps/opik-backend/src/test/java/com/comet/opik/api/resources/v1/priv/DatasetVersionResourceTest.java index 58a5ebe4acb..bb0e16f832b 100644 --- a/apps/opik-backend/src/test/java/com/comet/opik/api/resources/v1/priv/DatasetVersionResourceTest.java +++ b/apps/opik-backend/src/test/java/com/comet/opik/api/resources/v1/priv/DatasetVersionResourceTest.java @@ -3790,6 +3790,96 @@ void deleteDataset__whenVersionedDatasetWithTags__thenReturnNoContent() { @TestInstance(TestInstance.Lifecycle.PER_CLASS) class BatchVersioningTests { + @Test + @DisplayName("Success: Duplicate stable id in the version-creating batch counts once (OPIK-7891)") + void putItems__whenCreatingBatchRepeatsStableId__thenCountedOnce() { + var datasetId = createDataset(UUID.randomUUID().toString()); + + // The SDK's parallel upload sends every batch under one batch_group_id. The batch that + // arrives first CREATES the version, and that path derives itemsTotal from insertItems, + // which returns items.size() -- the raw list length, deliberately not a DB row count + // (ClickHouse async inserts report 0 before commit). A stable id repeated inside that + // first batch is therefore counted twice, while ClickHouse collapses it to one row. + var duplicatedId = TestIdGeneratorFactory.create().generateId(); + var items = List.of( + DatasetItem.builder() + .id(duplicatedId) + .source(DatasetItemSource.SDK) + .data(Map.of("value", JsonUtils.getJsonNodeFromString("\"first\""))) + .build(), + DatasetItem.builder() + .id(duplicatedId) + .source(DatasetItemSource.SDK) + .data(Map.of("value", JsonUtils.getJsonNodeFromString("\"second\""))) + .build(), + DatasetItem.builder() + .id(TestIdGeneratorFactory.create().generateId()) + .source(DatasetItemSource.SDK) + .data(Map.of("value", JsonUtils.getJsonNodeFromString("\"third\""))) + .build()); + + datasetResourceClient.createDatasetItems(DatasetItemBatch.builder() + .datasetId(datasetId) + .batchGroupId(UUID.randomUUID()) + .items(items) + .build(), TEST_WORKSPACE, API_KEY); + + var version = getLatestVersion(datasetId); + + // Two distinct ids went in, so the version holds two rows. + var stored = datasetResourceClient.getDatasetItems( + datasetId, 1, 100, version.versionHash(), API_KEY, TEST_WORKSPACE).content(); + assertThat(stored).hasSize(2); + + // items_total must agree with what is actually stored. + assertThat(version.itemsTotal()).isEqualTo(stored.size()); + } + + @Test + @DisplayName("Success: Duplicate stable id across batches in one group counts once (OPIK-7891)") + void putItems__whenDuplicateStableIdSpansBatchesInGroup__thenCountedOnce() { + var datasetId = createDataset(UUID.randomUUID().toString()); + var batchGroupId = UUID.randomUUID(); + + // Mirrors the SDK's parallel upload: several batches under one batch_group_id fold into + // one version. The first batch creates it, later batches append. A stable id present in + // both must contribute exactly one item to the total. + var sharedId = TestIdGeneratorFactory.create().generateId(); + + datasetResourceClient.createDatasetItems(DatasetItemBatch.builder() + .datasetId(datasetId) + .batchGroupId(batchGroupId) + .items(List.of( + DatasetItem.builder() + .id(sharedId) + .source(DatasetItemSource.SDK) + .data(Map.of("value", JsonUtils.getJsonNodeFromString("\"first\""))) + .build(), + DatasetItem.builder() + .id(TestIdGeneratorFactory.create().generateId()) + .source(DatasetItemSource.SDK) + .data(Map.of("value", JsonUtils.getJsonNodeFromString("\"other\""))) + .build())) + .build(), TEST_WORKSPACE, API_KEY); + + datasetResourceClient.createDatasetItems(DatasetItemBatch.builder() + .datasetId(datasetId) + .batchGroupId(batchGroupId) + .items(List.of(DatasetItem.builder() + .id(sharedId) + .source(DatasetItemSource.SDK) + .data(Map.of("value", JsonUtils.getJsonNodeFromString("\"updated\""))) + .build())) + .build(), TEST_WORKSPACE, API_KEY); + + var version = getLatestVersion(datasetId); + var stored = datasetResourceClient.getDatasetItems( + datasetId, 1, 100, version.versionHash(), API_KEY, TEST_WORKSPACE).content(); + + assertThat(stored).hasSize(2); + assertThat(version.itemsTotal()).isEqualTo(stored.size()); + } + @Test @DisplayName("Success: Multiple INSERT batches with same batch_group_id create single version") void putItems_whenSameBatchId_thenSingleVersion() { From 50aa19af7e0045ee4f28960f33a3ad57ee2cb5cb Mon Sep 17 00:00:00 2001 From: t Date: Mon, 24 Aug 2026 11:03:26 +0300 Subject: [PATCH 2/5] refactor(datasets): count the same id field the insert binds Drop the fall back to item.id() when datasetItemId is null. It looked defensive but was worse than nothing: the INSERT ten lines below binds item.datasetItemId().toString() with no null check, so a null cannot survive to be counted either way, and every caller normalizes datasetItemId before reaching here. All the fallback added was a second, divergent definition of item identity in the one place that must agree with what the INSERT writes. Addresses review feedback on #7966. Co-Authored-By: Claude Opus 5 (1M context) --- .../java/com/comet/opik/domain/DatasetItemVersionDAO.java | 7 ++++--- 1 file changed, 4 insertions(+), 3 deletions(-) diff --git a/apps/opik-backend/src/main/java/com/comet/opik/domain/DatasetItemVersionDAO.java b/apps/opik-backend/src/main/java/com/comet/opik/domain/DatasetItemVersionDAO.java index cedf2d7fd85..66a112ff0d0 100644 --- a/apps/opik-backend/src/main/java/com/comet/opik/domain/DatasetItemVersionDAO.java +++ b/apps/opik-backend/src/main/java/com/comet/opik/domain/DatasetItemVersionDAO.java @@ -3686,10 +3686,11 @@ public Mono insertItems(@NonNull UUID datasetId, @NonNull UUID newVersionI // // Count DISTINCT stable ids, not rows handed in (OPIK-7891): reads collapse a repeated // dataset_item_id to one row via LIMIT 1 BY, so counting the raw list inflates every - // version total derived from this value. Callers set datasetItemId before inserting; - // fall back to id so an unnormalized item still counts as itself rather than as null. + // version total derived from this value. Counting the same field the INSERT binds below + // keeps one definition of identity -- every caller normalizes datasetItemId first, and a + // null would fail at the bind regardless, so there is nothing to fall back to. long itemCount = items.stream() - .map(item -> item.datasetItemId() != null ? item.datasetItemId() : item.id()) + .map(DatasetItem::datasetItemId) .distinct() .count(); From 98b84826c0b71a8ea03f827fe002fb4e5e598c16 Mon Sep 17 00:00:00 2001 From: t Date: Wed, 26 Aug 2026 12:58:42 +0300 Subject: [PATCH 3/5] test(datasets): assert the surviving rows, not just how many Both OPIK-7891 tests checked stored.size() and left the row identities unasserted. On a de-duplication test that is the weakest possible check: the count is exactly what a bug is most likely to keep right. Keeping the wrong revision of the duplicate, or dropping the distinct row and keeping both duplicates, would each leave the size at 2 and pass. Now both name the rows that must survive and compare whole objects via the shared IGNORED_FIELDS_DATA_ITEM element comparator, so a wrong-winner bug fails. This also pins behaviour the tests previously only assumed: the last submitted revision of a repeated id is the one that survives. The itemsTotal assertion still ties the counter to stored.size() rather than a literal, which is only meaningful now that stored itself is pinned. Addresses review feedback on #7966. Co-Authored-By: Claude Opus 5 (1M context) --- .../v1/priv/DatasetVersionResourceTest.java | 70 +++++++++++++------ 1 file changed, 47 insertions(+), 23 deletions(-) diff --git a/apps/opik-backend/src/test/java/com/comet/opik/api/resources/v1/priv/DatasetVersionResourceTest.java b/apps/opik-backend/src/test/java/com/comet/opik/api/resources/v1/priv/DatasetVersionResourceTest.java index bb0e16f832b..aa76eb0c461 100644 --- a/apps/opik-backend/src/test/java/com/comet/opik/api/resources/v1/priv/DatasetVersionResourceTest.java +++ b/apps/opik-backend/src/test/java/com/comet/opik/api/resources/v1/priv/DatasetVersionResourceTest.java @@ -3801,22 +3801,31 @@ void putItems__whenCreatingBatchRepeatsStableId__thenCountedOnce() { // (ClickHouse async inserts report 0 before commit). A stable id repeated inside that // first batch is therefore counted twice, while ClickHouse collapses it to one row. var duplicatedId = TestIdGeneratorFactory.create().generateId(); + var distinctId = TestIdGeneratorFactory.create().generateId(); + + // The repeated id wins with its LAST submitted content: ClickHouse keeps one row per + // dataset_item_id and reads take the newest, so "second" survives and "first" does not. + var winningDuplicate = DatasetItem.builder() + .id(duplicatedId) + .datasetItemId(duplicatedId) + .source(DatasetItemSource.SDK) + .data(Map.of("value", JsonUtils.getJsonNodeFromString("\"second\""))) + .build(); + var distinctItem = DatasetItem.builder() + .id(distinctId) + .datasetItemId(distinctId) + .source(DatasetItemSource.SDK) + .data(Map.of("value", JsonUtils.getJsonNodeFromString("\"third\""))) + .build(); + var items = List.of( DatasetItem.builder() .id(duplicatedId) .source(DatasetItemSource.SDK) .data(Map.of("value", JsonUtils.getJsonNodeFromString("\"first\""))) .build(), - DatasetItem.builder() - .id(duplicatedId) - .source(DatasetItemSource.SDK) - .data(Map.of("value", JsonUtils.getJsonNodeFromString("\"second\""))) - .build(), - DatasetItem.builder() - .id(TestIdGeneratorFactory.create().generateId()) - .source(DatasetItemSource.SDK) - .data(Map.of("value", JsonUtils.getJsonNodeFromString("\"third\""))) - .build()); + winningDuplicate, + distinctItem); datasetResourceClient.createDatasetItems(DatasetItemBatch.builder() .datasetId(datasetId) @@ -3826,10 +3835,14 @@ void putItems__whenCreatingBatchRepeatsStableId__thenCountedOnce() { var version = getLatestVersion(datasetId); - // Two distinct ids went in, so the version holds two rows. + // Assert on the rows themselves, not just how many: a bug that kept the wrong revision + // of the duplicate, or dropped the distinct item and kept both duplicates, would leave + // the count at 2 and slip through a size-only check. var stored = datasetResourceClient.getDatasetItems( datasetId, 1, 100, version.versionHash(), API_KEY, TEST_WORKSPACE).content(); - assertThat(stored).hasSize(2); + assertThat(stored) + .usingRecursiveFieldByFieldElementComparatorIgnoringFields(IGNORED_FIELDS_DATA_ITEM) + .containsExactlyInAnyOrder(winningDuplicate, distinctItem); // items_total must agree with what is actually stored. assertThat(version.itemsTotal()).isEqualTo(stored.size()); @@ -3845,6 +3858,14 @@ void putItems__whenDuplicateStableIdSpansBatchesInGroup__thenCountedOnce() { // one version. The first batch creates it, later batches append. A stable id present in // both must contribute exactly one item to the total. var sharedId = TestIdGeneratorFactory.create().generateId(); + var otherId = TestIdGeneratorFactory.create().generateId(); + + var otherItem = DatasetItem.builder() + .id(otherId) + .datasetItemId(otherId) + .source(DatasetItemSource.SDK) + .data(Map.of("value", JsonUtils.getJsonNodeFromString("\"other\""))) + .build(); datasetResourceClient.createDatasetItems(DatasetItemBatch.builder() .datasetId(datasetId) @@ -3855,28 +3876,31 @@ void putItems__whenDuplicateStableIdSpansBatchesInGroup__thenCountedOnce() { .source(DatasetItemSource.SDK) .data(Map.of("value", JsonUtils.getJsonNodeFromString("\"first\""))) .build(), - DatasetItem.builder() - .id(TestIdGeneratorFactory.create().generateId()) - .source(DatasetItemSource.SDK) - .data(Map.of("value", JsonUtils.getJsonNodeFromString("\"other\""))) - .build())) + otherItem)) .build(), TEST_WORKSPACE, API_KEY); + // The second batch re-sends the shared id with new content, so it is an update: + // one row, holding the later revision. + var updatedShared = DatasetItem.builder() + .id(sharedId) + .datasetItemId(sharedId) + .source(DatasetItemSource.SDK) + .data(Map.of("value", JsonUtils.getJsonNodeFromString("\"updated\""))) + .build(); + datasetResourceClient.createDatasetItems(DatasetItemBatch.builder() .datasetId(datasetId) .batchGroupId(batchGroupId) - .items(List.of(DatasetItem.builder() - .id(sharedId) - .source(DatasetItemSource.SDK) - .data(Map.of("value", JsonUtils.getJsonNodeFromString("\"updated\""))) - .build())) + .items(List.of(updatedShared)) .build(), TEST_WORKSPACE, API_KEY); var version = getLatestVersion(datasetId); var stored = datasetResourceClient.getDatasetItems( datasetId, 1, 100, version.versionHash(), API_KEY, TEST_WORKSPACE).content(); - assertThat(stored).hasSize(2); + assertThat(stored) + .usingRecursiveFieldByFieldElementComparatorIgnoringFields(IGNORED_FIELDS_DATA_ITEM) + .containsExactlyInAnyOrder(updatedShared, otherItem); assertThat(version.itemsTotal()).isEqualTo(stored.size()); } From 23ac0174fb1ef023c6f30f7221d5afd4ce801130 Mon Sep 17 00:00:00 2001 From: t Date: Wed, 26 Aug 2026 13:09:04 +0300 Subject: [PATCH 4/5] [NA] [DOCS] docs: warn against bare hasSize in collection assertions hasSize alone asserts a count and nothing about identity, so a bug that preserves the count passes. Called out on #7966, where both new de-duplication tests checked stored.size() and left the row identities unasserted -- the count is exactly what a de-dup bug is most likely to get right. Co-Authored-By: Claude Opus 5 (1M context) --- .agents/skills/opik-backend/testing.md | 25 +++++++++++++++++++++++++ 1 file changed, 25 insertions(+) diff --git a/.agents/skills/opik-backend/testing.md b/.agents/skills/opik-backend/testing.md index 6627542c1cd..a07e4975bb1 100644 --- a/.agents/skills/opik-backend/testing.md +++ b/.agents/skills/opik-backend/testing.md @@ -284,6 +284,31 @@ assertThat(actual) `containsExactly` asserts size and content together — `hasSize` plus per-index checks does not, and lets an extra element through. +`hasSize` **on its own** is weaker still: it asserts a count and nothing about identity, so any +bug that preserves the count passes. This bites hardest on de-duplication, merge, and upsert +tests, where the count is exactly the thing a bug is most likely to keep right: + +```java +// ❌ BAD - passes if the wrong revision survived, or if the duplicate was kept +// and the distinct row dropped. Both keep the size at 2. +assertThat(stored).hasSize(2); +assertThat(version.itemsTotal()).isEqualTo(stored.size()); + +// ✅ GOOD - names the rows that must survive, so a wrong-winner bug fails +assertThat(stored) + .usingRecursiveFieldByFieldElementComparatorIgnoringFields(IGNORED_FIELDS_DATA_ITEM) + .containsExactlyInAnyOrder(winningDuplicate, distinctItem); +assertThat(version.itemsTotal()).isEqualTo(stored.size()); +``` + +Asserting a derived counter against `stored.size()` is good — it ties the counter to reality +rather than to a literal — but it is only as strong as the assertion on `stored` itself. Pin the +contents first, then tie the counter to them. + +Reuse the shared ignore-field constants (`IGNORED_FIELDS_DATA_ITEM` and friends) rather than +declaring a local list: they already encode which server-generated fields are not part of the +contract, and a local copy silently drifts from them. + These `containsExactly*` variants compare elements with the element type's own `equals`, which is what you want for exact-valued models. When the elements carry `BigDecimal` or `double`, the same exception that justifies a comparator on a single object applies per element — otherwise a From 77a681e24ed27f685dacc3d05e3b216293e6b29a Mon Sep 17 00:00:00 2001 From: t Date: Thu, 27 Aug 2026 19:19:43 +0300 Subject: [PATCH 5/5] test(datasets): drop the nondeterministic duplicate-winner assertion Every row in a batch is written with the same now64(9), and the read dedupes with ORDER BY dataset_item_id DESC, last_updated_at DESC LIMIT 1 BY dataset_item_id -- no tie-breaker. Which revision of a repeated id survives is therefore unspecified, so asserting that the later payload wins was pinning an accident. Assert what is guaranteed instead: exactly one row per distinct id, and the distinct item's content intact. Also corrects a comment that described insertItems' pre-fix return value in the present tense. Addresses review feedback on #7966. Co-Authored-By: Claude Opus 5 (1M context) --- .../v1/priv/DatasetVersionResourceTest.java | 43 +++++++++++-------- 1 file changed, 24 insertions(+), 19 deletions(-) diff --git a/apps/opik-backend/src/test/java/com/comet/opik/api/resources/v1/priv/DatasetVersionResourceTest.java b/apps/opik-backend/src/test/java/com/comet/opik/api/resources/v1/priv/DatasetVersionResourceTest.java index aa76eb0c461..616adfcae59 100644 --- a/apps/opik-backend/src/test/java/com/comet/opik/api/resources/v1/priv/DatasetVersionResourceTest.java +++ b/apps/opik-backend/src/test/java/com/comet/opik/api/resources/v1/priv/DatasetVersionResourceTest.java @@ -3796,21 +3796,14 @@ void putItems__whenCreatingBatchRepeatsStableId__thenCountedOnce() { var datasetId = createDataset(UUID.randomUUID().toString()); // The SDK's parallel upload sends every batch under one batch_group_id. The batch that - // arrives first CREATES the version, and that path derives itemsTotal from insertItems, - // which returns items.size() -- the raw list length, deliberately not a DB row count - // (ClickHouse async inserts report 0 before commit). A stable id repeated inside that - // first batch is therefore counted twice, while ClickHouse collapses it to one row. + // arrives first CREATES the version, and that path derives itemsTotal from insertItems. + // That used to return items.size() -- the raw list length, deliberately not a DB row count + // (ClickHouse async inserts report 0 before commit) -- so a stable id repeated inside the + // first batch was counted twice while ClickHouse collapsed it to one row. It now counts + // distinct dataset_item_ids, which is what this test pins. var duplicatedId = TestIdGeneratorFactory.create().generateId(); var distinctId = TestIdGeneratorFactory.create().generateId(); - // The repeated id wins with its LAST submitted content: ClickHouse keeps one row per - // dataset_item_id and reads take the newest, so "second" survives and "first" does not. - var winningDuplicate = DatasetItem.builder() - .id(duplicatedId) - .datasetItemId(duplicatedId) - .source(DatasetItemSource.SDK) - .data(Map.of("value", JsonUtils.getJsonNodeFromString("\"second\""))) - .build(); var distinctItem = DatasetItem.builder() .id(distinctId) .datasetItemId(distinctId) @@ -3824,7 +3817,11 @@ void putItems__whenCreatingBatchRepeatsStableId__thenCountedOnce() { .source(DatasetItemSource.SDK) .data(Map.of("value", JsonUtils.getJsonNodeFromString("\"first\""))) .build(), - winningDuplicate, + DatasetItem.builder() + .id(duplicatedId) + .source(DatasetItemSource.SDK) + .data(Map.of("value", JsonUtils.getJsonNodeFromString("\"second\""))) + .build(), distinctItem); datasetResourceClient.createDatasetItems(DatasetItemBatch.builder() @@ -3834,15 +3831,23 @@ void putItems__whenCreatingBatchRepeatsStableId__thenCountedOnce() { .build(), TEST_WORKSPACE, API_KEY); var version = getLatestVersion(datasetId); - - // Assert on the rows themselves, not just how many: a bug that kept the wrong revision - // of the duplicate, or dropped the distinct item and kept both duplicates, would leave - // the count at 2 and slip through a size-only check. var stored = datasetResourceClient.getDatasetItems( datasetId, 1, 100, version.versionHash(), API_KEY, TEST_WORKSPACE).content(); + + // Which revision of the repeated id survives is deliberately NOT asserted: every row in a + // batch is written with the same now64(9), and the read dedupes with + // `ORDER BY dataset_item_id DESC, last_updated_at DESC LIMIT 1 BY dataset_item_id` -- no + // tie-breaker, so either payload may win. Pinning "second" would be asserting an accident. + // What is guaranteed, and what the fix is about: exactly one row per distinct id, and a + // counter that agrees with it. + assertThat(stored).extracting(DatasetItem::id) + .containsExactlyInAnyOrder(duplicatedId, distinctId); assertThat(stored) - .usingRecursiveFieldByFieldElementComparatorIgnoringFields(IGNORED_FIELDS_DATA_ITEM) - .containsExactlyInAnyOrder(winningDuplicate, distinctItem); + .filteredOn(item -> distinctId.equals(item.id())) + .singleElement() + .usingRecursiveComparison() + .ignoringFields(IGNORED_FIELDS_DATA_ITEM) + .isEqualTo(distinctItem); // items_total must agree with what is actually stored. assertThat(version.itemsTotal()).isEqualTo(stored.size());