[OPIK-7891] [BE] fix: count distinct item ids when sizing a dataset version

.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

apps/opik-backend/src/main/java/com/comet/opik/domain/DatasetItemVersionDAO.java

@@ -3683,7 +3683,16 @@ public Mono<Long> 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");

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() {