Skip to content

Commit d863779

Browse files
JetoPistolaJetoPistola
authored
Merge branch 'main' into danield/OPIK-8181-centralise-dataset-item-test-assertions
2 parents c79b120 + 0b904ae commit d863779

55 files changed

Lines changed: 2330 additions & 366 deletions

File tree

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.

‎.github/scripts/discover-backend-tests.sh‎

Lines changed: 8 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -4,6 +4,12 @@ set -euo pipefail
44
NUM_GROUPS=16
55
UNIT_TIMEOUT=10
66
INTEGRATION_TIMEOUT=20
7+
# Per-test timeout handed to JUnit, overriding the default in junit-platform.properties.
8+
# CI reruns failures 3 times, so a deterministic hang costs 4x these values: 8m against the 10m
9+
# unit wall, 16m against the 20m integration wall. Both must stay under their job timeout above,
10+
# or the job is cancelled before Surefire can name the offending test.
11+
UNIT_TEST_TIMEOUT=2m
12+
INTEGRATION_TEST_TIMEOUT=4m
713
TEST_DIR="src/test/java"
814
PATTERN="DropwizardAppExtensionProvider\|MySQLContainer\|ClickHouseContainer\|RedisContainer\|MinIOContainer"
915

@@ -62,9 +68,9 @@ done
6268

6369
# Build JSON matrix: unit tests + N integration groups
6470
matrix="{\"include\":["
65-
matrix+="{\"name\":\"Unit Tests\",\"tests\":\"$unit_list\",\"timeout\":$UNIT_TIMEOUT}"
71+
matrix+="{\"name\":\"Unit Tests\",\"tests\":\"$unit_list\",\"timeout\":$UNIT_TIMEOUT,\"testTimeout\":\"$UNIT_TEST_TIMEOUT\"}"
6672
for ((i=1; i<=NUM_GROUPS; i++)); do
67-
matrix+=",{\"name\":\"Integration Group $i\",\"tests\":\"${group_list[$i]}\",\"timeout\":$INTEGRATION_TIMEOUT}"
73+
matrix+=",{\"name\":\"Integration Group $i\",\"tests\":\"${group_list[$i]}\",\"timeout\":$INTEGRATION_TIMEOUT,\"testTimeout\":\"$INTEGRATION_TEST_TIMEOUT\"}"
6874
done
6975
matrix+="]}"
7076
echo "matrix=$matrix" >> "$GITHUB_OUTPUT"

‎.github/workflows/backend_tests.yml‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -84,6 +84,7 @@ jobs:
8484
-Dtest="${{ matrix.tests }}"
8585
-Dmaven.test.failure.ignore=true
8686
-Dsurefire.rerunFailingTestsCount=3
87+
-Djunit.jupiter.execution.timeout.testable.method.default="${{ matrix.testTimeout }}"
8788
8889
- name: Publish Test Report
8990
uses: EnricoMi/publish-unit-test-result-action/linux@v2

‎.github/workflows/test_docs_links.yml‎

Lines changed: 0 additions & 65 deletions
This file was deleted.

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

Lines changed: 53 additions & 31 deletions
Original file line numberDiff line numberDiff line change
@@ -560,7 +560,12 @@ public DatasetPage find(int page, int size, @NonNull DatasetCriteria criteria, L
560560
}
561561

562562
// For now, we are not going to use the criteria.withExperimentsOnly() method due to the migration.
563-
return template.inTransaction(READ_ONLY, handle -> {
563+
// Enrichment runs outside the transaction so the MySQL connection is not held open across the three
564+
// ClickHouse round-trips it makes. Trade-off: the page rows and the dataset versions were previously
565+
// read in one READ_ONLY transaction and are now two, so a version written between them is observable
566+
// where it was not before. Benign for count/summary metadata, and only find() is affected -- findById,
567+
// findByNameDetailed and the experiments-only branch already enriched outside any transaction.
568+
DatasetPage rawPage = template.inTransaction(READ_ONLY, handle -> {
564569

565570
var repository = handle.attach(DatasetDAO.class);
566571
int offset = (page - 1) * size;
@@ -569,14 +574,17 @@ public DatasetPage find(int page, int size, @NonNull DatasetCriteria criteria, L
569574
criteria.withExperimentsOnly(),
570575
criteria.withOptimizationsOnly(), visibility, filtersSQL, filterMapping);
571576

572-
List<Dataset> datasets = enrichDatasetWithAdditionalInformation(
573-
repository.find(size, offset, workspaceId, criteria.name(), criteria.projectId(),
574-
criteria.withExperimentsOnly(),
575-
criteria.withOptimizationsOnly(),
576-
sortingFieldsSql, visibility, filtersSQL, filterMapping));
577+
List<Dataset> datasets = repository.find(size, offset, workspaceId, criteria.name(), criteria.projectId(),
578+
criteria.withExperimentsOnly(),
579+
criteria.withOptimizationsOnly(),
580+
sortingFieldsSql, visibility, filtersSQL, filterMapping);
577581

578582
return new DatasetPage(datasets, page, datasets.size(), count, sortingFactory.getSortableFields());
579583
});
584+
585+
List<Dataset> datasets = enrichDatasetWithAdditionalInformation(rawPage.content());
586+
587+
return new DatasetPage(datasets, page, datasets.size(), rawPage.total(), sortingFactory.getSortableFields());
580588
}
581589

582590
private Mono<DatasetPage> fetchUsingTempTable(int page, int size, DatasetCriteria criteria, Set<UUID> ids,
@@ -708,31 +716,47 @@ public List<UUID> findIdsByPartialName(@NonNull String workspaceId, @NonNull Str
708716
private List<Dataset> enrichDatasetWithAdditionalInformation(List<Dataset> datasets) {
709717
Set<UUID> ids = datasets.stream().map(Dataset::id).collect(toSet());
710718

711-
Map<UUID, ExperimentSummary> experimentSummary = experimentItemDAO.findExperimentSummaryByDatasetIds(ids)
712-
.contextWrite(ctx -> AsyncUtils.setRequestContext(ctx, requestContext))
713-
.toStream()
714-
.collect(toMap(ExperimentSummary::datasetId, Function.identity()));
715-
716-
Map<UUID, DatasetItemSummary> datasetItemSummaryMap = datasetItemDAO.findDatasetItemSummaryByDatasetIds(ids)
717-
.contextWrite(ctx -> AsyncUtils.setRequestContext(ctx, requestContext))
718-
.toStream()
719-
.collect(toMap(DatasetItemSummary::datasetId, Function.identity()));
719+
if (ids.isEmpty()) {
720+
return datasets;
721+
}
720722

721-
Map<UUID, OptimizationDAO.OptimizationSummary> optimizationSummaryMap = optimizationDAO
722-
.findOptimizationSummaryByDatasetIds(ids)
723-
.contextWrite(ctx -> AsyncUtils.setRequestContext(ctx, requestContext))
724-
.toStream()
725-
.collect(toMap(OptimizationDAO.OptimizationSummary::datasetId, Function.identity()));
723+
// RequestContext is @RequestScoped and therefore thread-bound: resolve it here, on the request thread,
724+
// so the concurrent subscriptions below never call requestContext.get() from a worker thread.
725+
String workspaceId = requestContext.get().getWorkspaceId();
726+
String userName = requestContext.get().getUserName();
726727

727-
Map<UUID, DatasetVersion> latestVersionsByDatasetId = fetchLatestVersionsByDatasetIds(ids);
728+
// collect(...) with Collectors.toMap rather than Flux.collectMap: collectMap is last-wins, whereas the
729+
// serial code this replaces threw on a duplicate dataset_id. All three queries GROUP BY dataset_id so
730+
// duplicates should not occur; keeping the loud form means a query change that broke that assumption
731+
// fails instead of silently dropping one row's summary.
732+
var enrichmentData = Mono.zip(
733+
experimentItemDAO.findExperimentSummaryByDatasetIds(ids)
734+
.collect(toMap(ExperimentSummary::datasetId, Function.identity())),
735+
datasetItemDAO.findDatasetItemSummaryByDatasetIds(ids)
736+
.collect(toMap(DatasetItemSummary::datasetId, Function.identity())),
737+
optimizationDAO.findOptimizationSummaryByDatasetIds(ids)
738+
.collect(toMap(OptimizationDAO.OptimizationSummary::datasetId, Function.identity())),
739+
// defaultIfEmpty guards the zip: a Mono that completes empty makes zip emit nothing at all,
740+
// which would turn an absent-versions result into a null and NPE below.
741+
Mono.fromCallable(() -> fetchLatestVersionsByDatasetIds(ids, workspaceId))
742+
.subscribeOn(Schedulers.boundedElastic())
743+
.defaultIfEmpty(Map.of()))
744+
.contextWrite(ctx -> AsyncUtils.setRequestContext(ctx, userName, workspaceId))
745+
.block();
746+
747+
Map<UUID, ExperimentSummary> experimentSummaryMap = enrichmentData.getT1();
748+
Map<UUID, DatasetItemSummary> datasetItemSummaryMap = enrichmentData.getT2();
749+
Map<UUID, OptimizationDAO.OptimizationSummary> optimizationSummaryMap = enrichmentData.getT3();
750+
Map<UUID, DatasetVersion> latestVersionsByDatasetId = enrichmentData.getT4();
728751

729752
return datasets.stream()
730753
.map(dataset -> {
731-
var resume = experimentSummary.computeIfAbsent(dataset.id(), ExperimentSummary::empty);
732-
var datasetItemSummary = datasetItemSummaryMap.computeIfAbsent(dataset.id(),
733-
DatasetItemSummary::empty);
734-
var optimizationSummary = optimizationSummaryMap.computeIfAbsent(dataset.id(),
735-
OptimizationDAO.OptimizationSummary::empty);
754+
var experimentSummary = experimentSummaryMap.getOrDefault(dataset.id(),
755+
ExperimentSummary.empty(dataset.id()));
756+
var datasetItemSummary = datasetItemSummaryMap.getOrDefault(dataset.id(),
757+
DatasetItemSummary.empty(dataset.id()));
758+
var optimizationSummary = optimizationSummaryMap.getOrDefault(dataset.id(),
759+
OptimizationDAO.OptimizationSummary.empty(dataset.id()));
736760
var latestVersion = latestVersionsByDatasetId.get(dataset.id());
737761

738762
// When versioning is enabled and a latest version exists, use itemsTotal from the version
@@ -746,24 +770,22 @@ private List<Dataset> enrichDatasetWithAdditionalInformation(List<Dataset> datas
746770
}
747771

748772
return dataset.toBuilder()
749-
.experimentCount(resume.experimentCount())
773+
.experimentCount(experimentSummary.experimentCount())
750774
.datasetItemsCount(itemsCount)
751775
.optimizationCount(optimizationSummary.optimizationCount())
752-
.mostRecentExperimentAt(resume.mostRecentExperimentAt())
776+
.mostRecentExperimentAt(experimentSummary.mostRecentExperimentAt())
753777
.mostRecentOptimizationAt(optimizationSummary.mostRecentOptimizationAt())
754778
.latestVersion(DatasetVersionMapper.INSTANCE.toDatasetVersionSummary(latestVersion))
755779
.build();
756780
})
757781
.toList();
758782
}
759783

760-
private Map<UUID, DatasetVersion> fetchLatestVersionsByDatasetIds(Set<UUID> datasetIds) {
784+
private Map<UUID, DatasetVersion> fetchLatestVersionsByDatasetIds(Set<UUID> datasetIds, String workspaceId) {
761785
if (datasetIds.isEmpty()) {
762786
return Map.of();
763787
}
764788

765-
String workspaceId = requestContext.get().getWorkspaceId();
766-
767789
return template.inTransaction(READ_ONLY, handle -> {
768790
var dao = handle.attach(DatasetVersionDAO.class);
769791
List<DatasetVersion> latestVersions = dao.findLatestVersionsByDatasetIds(datasetIds, workspaceId);

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

Lines changed: 8 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -170,8 +170,10 @@ AND notEquals(start_time, toDateTime64('1970-01-01 00:00:00.000', 9)),
170170
AND workspace_id = :workspace_id
171171
AND id >= :uuid_from_time
172172
AND id \\<= :uuid_to_time
173-
AND toMonday(id_at) >= toMonday(UUIDv7ToDateTime(toUUID(:uuid_from_time), 'UTC'))
174-
AND toMonday(id_at) \\<= toMonday(UUIDv7ToDateTime(toUUID(:uuid_to_time), 'UTC'))
173+
AND (toDate32(id_at) - toIntervalDay(toDayOfWeek(id_at, 1)))
174+
>= (toDate32(UUIDv7ToDateTime(toUUID(:uuid_from_time), 'UTC')) - toIntervalDay(toDayOfWeek(UUIDv7ToDateTime(toUUID(:uuid_from_time), 'UTC'), 1)))
175+
AND (toDate32(id_at) - toIntervalDay(toDayOfWeek(id_at, 1)))
176+
\\<= (toDate32(UUIDv7ToDateTime(toUUID(:uuid_to_time), 'UTC')) - toIntervalDay(toDayOfWeek(UUIDv7ToDateTime(toUUID(:uuid_to_time), 'UTC'), 1)))
175177
<if(trace_filters)> AND <trace_filters> <endif>
176178
<if(trace_feedback_scores_filters)>
177179
AND id in (
@@ -364,8 +366,10 @@ WITH traces_final AS (
364366
WHERE workspace_id = :workspace_id
365367
AND project_id = :project_id
366368
AND id >= :uuid_from_time AND id \\<= :uuid_to_time
367-
AND toMonday(id_at) >= toMonday(UUIDv7ToDateTime(toUUID(:uuid_from_time), 'UTC'))
368-
AND toMonday(id_at) \\<= toMonday(UUIDv7ToDateTime(toUUID(:uuid_to_time), 'UTC'))
369+
AND (toDate32(id_at) - toIntervalDay(toDayOfWeek(id_at, 1)))
370+
>= (toDate32(UUIDv7ToDateTime(toUUID(:uuid_from_time), 'UTC')) - toIntervalDay(toDayOfWeek(UUIDv7ToDateTime(toUUID(:uuid_from_time), 'UTC'), 1)))
371+
AND (toDate32(id_at) - toIntervalDay(toDayOfWeek(id_at, 1)))
372+
\\<= (toDate32(UUIDv7ToDateTime(toUUID(:uuid_to_time), 'UTC')) - toIntervalDay(toDayOfWeek(UUIDv7ToDateTime(toUUID(:uuid_to_time), 'UTC'), 1)))
369373
AND thread_id \\<> ''
370374
), trace_threads_final AS (
371375
SELECT

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

Lines changed: 12 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -281,9 +281,11 @@ traces_filtered AS (
281281
WHERE project_id = :project_id
282282
AND workspace_id = :workspace_id
283283
<if(uuid_from_time)> AND id >= :uuid_from_time
284-
AND toMonday(id_at) >= toMonday(UUIDv7ToDateTime(toUUID(:uuid_from_time), 'UTC'))<endif>
284+
AND (toDate32(id_at) - toIntervalDay(toDayOfWeek(id_at, 1)))
285+
>= (toDate32(UUIDv7ToDateTime(toUUID(:uuid_from_time), 'UTC')) - toIntervalDay(toDayOfWeek(UUIDv7ToDateTime(toUUID(:uuid_from_time), 'UTC'), 1)))<endif>
285286
<if(uuid_to_time)> AND id \\<= :uuid_to_time
286-
AND toMonday(id_at) \\<= toMonday(UUIDv7ToDateTime(toUUID(:uuid_to_time), 'UTC'))<endif>
287+
AND (toDate32(id_at) - toIntervalDay(toDayOfWeek(id_at, 1)))
288+
\\<= (toDate32(UUIDv7ToDateTime(toUUID(:uuid_to_time), 'UTC')) - toIntervalDay(toDayOfWeek(UUIDv7ToDateTime(toUUID(:uuid_to_time), 'UTC'), 1)))<endif>
287289
<if(trace_filters)> AND <trace_filters> <endif>
288290
<if(trace_feedback_scores_filters)>
289291
AND id in (
@@ -957,9 +959,11 @@ TO toDateTime(UUIDv7ToDateTime(toUUID(:uuid_to_time)))
957959
WHERE workspace_id = :workspace_id
958960
<if(project_ids)> AND project_id IN :project_ids <endif>
959961
<if(uuid_from_time)>AND id >= :uuid_from_time
960-
AND toMonday(id_at) >= toMonday(UUIDv7ToDateTime(toUUID(:uuid_from_time), 'UTC'))<endif>
962+
AND (toDate32(id_at) - toIntervalDay(toDayOfWeek(id_at, 1)))
963+
>= (toDate32(UUIDv7ToDateTime(toUUID(:uuid_from_time), 'UTC')) - toIntervalDay(toDayOfWeek(UUIDv7ToDateTime(toUUID(:uuid_from_time), 'UTC'), 1)))<endif>
961964
<if(uuid_to_time)>AND id \\<= :uuid_to_time
962-
AND toMonday(id_at) \\<= toMonday(UUIDv7ToDateTime(toUUID(:uuid_to_time), 'UTC'))<endif>
965+
AND (toDate32(id_at) - toIntervalDay(toDayOfWeek(id_at, 1)))
966+
\\<= (toDate32(UUIDv7ToDateTime(toUUID(:uuid_to_time), 'UTC')) - toIntervalDay(toDayOfWeek(UUIDv7ToDateTime(toUUID(:uuid_to_time), 'UTC'), 1)))<endif>
963967
SETTINGS log_comment = '<log_comment>';
964968
""";
965969

@@ -971,9 +975,11 @@ AND toMonday(id_at) \\<= toMonday(UUIDv7ToDateTime(toUUID(:uuid_to_time), 'UTC')
971975
AND length(error_info) > 0
972976
<if(project_ids)> AND project_id IN :project_ids <endif>
973977
<if(uuid_from_time)>AND id >= :uuid_from_time
974-
AND toMonday(id_at) >= toMonday(UUIDv7ToDateTime(toUUID(:uuid_from_time), 'UTC'))<endif>
978+
AND (toDate32(id_at) - toIntervalDay(toDayOfWeek(id_at, 1)))
979+
>= (toDate32(UUIDv7ToDateTime(toUUID(:uuid_from_time), 'UTC')) - toIntervalDay(toDayOfWeek(UUIDv7ToDateTime(toUUID(:uuid_from_time), 'UTC'), 1)))<endif>
975980
<if(uuid_to_time)>AND id \\<= :uuid_to_time
976-
AND toMonday(id_at) \\<= toMonday(UUIDv7ToDateTime(toUUID(:uuid_to_time), 'UTC'))<endif>
981+
AND (toDate32(id_at) - toIntervalDay(toDayOfWeek(id_at, 1)))
982+
\\<= (toDate32(UUIDv7ToDateTime(toUUID(:uuid_to_time), 'UTC')) - toIntervalDay(toDayOfWeek(UUIDv7ToDateTime(toUUID(:uuid_to_time), 'UTC'), 1)))<endif>
977983
SETTINGS log_comment = '<log_comment>';
978984
""";
979985

0 commit comments

Comments
 (0)