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) <noreply@anthropic.com>

Author: petros-double-test1

apps/opik-backend/src/test/java/com/comet/opik/api/resources/v1/priv/DatasetVersionResourceTest.java

@@ -3568,22 +3568,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)
@@ -3593,10 +3602,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());
@@ -3612,6 +3625,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)
@@ -3622,28 +3643,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());
         }