Skip to content

Commit c37e5c6

Browse files
petros-double-test1claude
petros-double-test1
and
claude
committed
[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) <noreply@anthropic.com>
1 parent 7ad36cf commit c37e5c6

2 files changed

Lines changed: 99 additions & 1 deletion

File tree

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

Lines changed: 9 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -3499,7 +3499,15 @@ public Mono<Long> insertItems(@NonNull UUID datasetId, @NonNull UUID newVersionI
34993499

35003500
// Note: ClickHouse with async inserts returns 0 immediately before commit.
35013501
// We return the count of items we're inserting instead of relying on getRowsUpdated.
3502-
long itemCount = items.size();
3502+
//
3503+
// Count DISTINCT stable ids, not rows handed in (OPIK-7891): reads collapse a repeated
3504+
// dataset_item_id to one row via LIMIT 1 BY, so counting the raw list inflates every
3505+
// version total derived from this value. Callers set datasetItemId before inserting;
3506+
// fall back to id so an unnormalized item still counts as itself rather than as null.
3507+
long itemCount = items.stream()
3508+
.map(item -> item.datasetItemId() != null ? item.datasetItemId() : item.id())
3509+
.distinct()
3510+
.count();
35033511

35043512
return asyncTemplate.nonTransaction(connection -> {
35053513
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‎

Lines changed: 90 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -3557,6 +3557,96 @@ void deleteDataset__whenVersionedDatasetWithTags__thenReturnNoContent() {
35573557
@TestInstance(TestInstance.Lifecycle.PER_CLASS)
35583558
class BatchVersioningTests {
35593559

3560+
@Test
3561+
@DisplayName("Success: Duplicate stable id in the version-creating batch counts once (OPIK-7891)")
3562+
void putItems__whenCreatingBatchRepeatsStableId__thenCountedOnce() {
3563+
var datasetId = createDataset(UUID.randomUUID().toString());
3564+
3565+
// The SDK's parallel upload sends every batch under one batch_group_id. The batch that
3566+
// arrives first CREATES the version, and that path derives itemsTotal from insertItems,
3567+
// which returns items.size() -- the raw list length, deliberately not a DB row count
3568+
// (ClickHouse async inserts report 0 before commit). A stable id repeated inside that
3569+
// first batch is therefore counted twice, while ClickHouse collapses it to one row.
3570+
var duplicatedId = TestIdGeneratorFactory.create().generateId();
3571+
var items = List.of(
3572+
DatasetItem.builder()
3573+
.id(duplicatedId)
3574+
.source(DatasetItemSource.SDK)
3575+
.data(Map.of("value", JsonUtils.getJsonNodeFromString("\"first\"")))
3576+
.build(),
3577+
DatasetItem.builder()
3578+
.id(duplicatedId)
3579+
.source(DatasetItemSource.SDK)
3580+
.data(Map.of("value", JsonUtils.getJsonNodeFromString("\"second\"")))
3581+
.build(),
3582+
DatasetItem.builder()
3583+
.id(TestIdGeneratorFactory.create().generateId())
3584+
.source(DatasetItemSource.SDK)
3585+
.data(Map.of("value", JsonUtils.getJsonNodeFromString("\"third\"")))
3586+
.build());
3587+
3588+
datasetResourceClient.createDatasetItems(DatasetItemBatch.builder()
3589+
.datasetId(datasetId)
3590+
.batchGroupId(UUID.randomUUID())
3591+
.items(items)
3592+
.build(), TEST_WORKSPACE, API_KEY);
3593+
3594+
var version = getLatestVersion(datasetId);
3595+
3596+
// Two distinct ids went in, so the version holds two rows.
3597+
var stored = datasetResourceClient.getDatasetItems(
3598+
datasetId, 1, 100, version.versionHash(), API_KEY, TEST_WORKSPACE).content();
3599+
assertThat(stored).hasSize(2);
3600+
3601+
// items_total must agree with what is actually stored.
3602+
assertThat(version.itemsTotal()).isEqualTo(stored.size());
3603+
}
3604+
3605+
@Test
3606+
@DisplayName("Success: Duplicate stable id across batches in one group counts once (OPIK-7891)")
3607+
void putItems__whenDuplicateStableIdSpansBatchesInGroup__thenCountedOnce() {
3608+
var datasetId = createDataset(UUID.randomUUID().toString());
3609+
var batchGroupId = UUID.randomUUID();
3610+
3611+
// Mirrors the SDK's parallel upload: several batches under one batch_group_id fold into
3612+
// one version. The first batch creates it, later batches append. A stable id present in
3613+
// both must contribute exactly one item to the total.
3614+
var sharedId = TestIdGeneratorFactory.create().generateId();
3615+
3616+
datasetResourceClient.createDatasetItems(DatasetItemBatch.builder()
3617+
.datasetId(datasetId)
3618+
.batchGroupId(batchGroupId)
3619+
.items(List.of(
3620+
DatasetItem.builder()
3621+
.id(sharedId)
3622+
.source(DatasetItemSource.SDK)
3623+
.data(Map.of("value", JsonUtils.getJsonNodeFromString("\"first\"")))
3624+
.build(),
3625+
DatasetItem.builder()
3626+
.id(TestIdGeneratorFactory.create().generateId())
3627+
.source(DatasetItemSource.SDK)
3628+
.data(Map.of("value", JsonUtils.getJsonNodeFromString("\"other\"")))
3629+
.build()))
3630+
.build(), TEST_WORKSPACE, API_KEY);
3631+
3632+
datasetResourceClient.createDatasetItems(DatasetItemBatch.builder()
3633+
.datasetId(datasetId)
3634+
.batchGroupId(batchGroupId)
3635+
.items(List.of(DatasetItem.builder()
3636+
.id(sharedId)
3637+
.source(DatasetItemSource.SDK)
3638+
.data(Map.of("value", JsonUtils.getJsonNodeFromString("\"updated\"")))
3639+
.build()))
3640+
.build(), TEST_WORKSPACE, API_KEY);
3641+
3642+
var version = getLatestVersion(datasetId);
3643+
var stored = datasetResourceClient.getDatasetItems(
3644+
datasetId, 1, 100, version.versionHash(), API_KEY, TEST_WORKSPACE).content();
3645+
3646+
assertThat(stored).hasSize(2);
3647+
assertThat(version.itemsTotal()).isEqualTo(stored.size());
3648+
}
3649+
35603650
@Test
35613651
@DisplayName("Success: Multiple INSERT batches with same batch_group_id create single version")
35623652
void putItems_whenSameBatchId_thenSingleVersion() {

0 commit comments

Comments
 (0)