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 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..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 @@ -3683,7 +3683,16 @@ 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. 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(DatasetItem::datasetItemId) + .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..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 @@ -3790,6 +3790,125 @@ 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. + // 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(); + + 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(), + distinctItem); + + datasetResourceClient.createDatasetItems(DatasetItemBatch.builder() + .datasetId(datasetId) + .batchGroupId(UUID.randomUUID()) + .items(items) + .build(), TEST_WORKSPACE, API_KEY); + + var version = getLatestVersion(datasetId); + 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) + .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()); + } + + @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(); + 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) + .batchGroupId(batchGroupId) + .items(List.of( + DatasetItem.builder() + .id(sharedId) + .source(DatasetItemSource.SDK) + .data(Map.of("value", JsonUtils.getJsonNodeFromString("\"first\""))) + .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(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) + .usingRecursiveFieldByFieldElementComparatorIgnoringFields(IGNORED_FIELDS_DATA_ITEM) + .containsExactlyInAnyOrder(updatedShared, otherItem); + assertThat(version.itemsTotal()).isEqualTo(stored.size()); + } + @Test @DisplayName("Success: Multiple INSERT batches with same batch_group_id create single version") void putItems_whenSameBatchId_thenSingleVersion() {