-
[OPIK-4383] [BE] Redis stream subscriber for debounced experiment aggregates recomputation (#5371)
frostbyte_neo 发布于
2026-03-09 16:54:39 +00:00 - [OPIK-4380] [BE] Add experiment aggregates for denormalized metrics
- Add experiment_aggregates and experiment_item_aggregates tables
- Implement ExperimentAggregatesDAO with population and query methods
- Add ExperimentAggregatesService for aggregation management
- Refactor DTOs into organized model classes:
- ExperimentAggregatesModel: aggregation results
- ExperimentEntityData: entity models
- ExperimentSourceData: raw source data
- ExperimentAggregatesUtils: utilities
- Add FEEDBACK_SCORES_AGGREGATED filter strategy for map-based filtering
- Add comprehensive integration tests (10/10 passing)
- Configure batch size and parallelism settings
- [OPIK-4380] [BE] Add MySQL deadlock retry mechanism for concurrent dataset operations
Problem:
- MySQL deadlock on dataset_version_tags composite PRIMARY KEY (workspace_id, dataset_id, tag)
- Occurred during parallel dataset creation in same workspace
- Multiple threads inserting "latest" tag for different datasets caused lock contention
- Experiments with parallel execution were failing with MySQLTransactionRollbackException
Solution:
- Add handleOnDeadLocks() method in RetryUtils with:
- 5 retry attempts with exponential backoff (250ms to 2s)
- 0.5 jitter to reduce thundering herd effect
- Recursive isDatabaseDeadlock() detection for MySQLTransactionRollbackException
- Apply retry logic in DatasetItemService.setDatasetItemVersion()
- Enables concurrent dataset creation for same workspace
Impact:
- Supports parallel experiment execution with proper deadlock handling
- Test success rate improved from 0/10 to 10/10 in ExperimentAggregatesIntegrationTest
-
Fix visibility
-
[OPIK-4380] [BE] Address PR review comments for experiment aggregates
Fixed 11 automated review comments from baz-reviewer:
CRITICAL fixes:
- Prevent NPE on null span aggregations by adding coalesce() in SQL
- Handle multi-project experiments with LIMIT 1 in GET_PROJECT_ID
- Handle zero-item experiments with empty aggregation helpers
- Bind feedback_scores_percentiles map instead of empty CAST
HIGH priority fixes:
- Use toDecimal128(12) instead of toDecimal64(9) for cost percentiles
- Add null-safe tags handling with Optional.ofNullable()
- Include exception objects in retry logging for stack traces
MEDIUM priority fixes:
- Add missing log_comment to SELECT_EXPERIMENT_BY_ID query
- Add missing log_comment to GET_PROJECT_ID query
LOW priority fixes:
- Remove duplicate "id" binding in bindItemsParameters
- Enhance batchSize config documentation with details
All 11 integration tests passing.
- [OPIK-4382] [BE] Refactor experiment aggregates with import cleanup and Optional patterns
- Add missing imports for IntStream, ProjectStats, and other dependencies
- Replace fully-qualified class names with proper imports across DAO and Service classes
- Fix IS_NOT_EMPTY filter handling for FEEDBACK_SCORES_AGGREGATED strategies
- Refactor null checks to use Optional in mapping methods:
- mapFeedbackScoreAggregations, mapExperimentFromAggregates
- mapFeedbackScoreData, mapExperimentGroupAggregationItem
- Batch insert preparation with Optional chains
- Improve code readability and maintainability with functional patterns
-
[OPIK-4380] [BE] Fix table definition
-
[OPIK-4380] [BE] Address PR comments and consolidate DatasetItemService methods
- Fix tags NPE in ExperimentAggregatesDAO with defaultIfNull
- Remove unnecessary FINAL clause from GET_EXPERIMENT_DATA query
- Fix test naming in ExperimentAggregatesIntegrationTest
- Consolidate 7 duplicate createVersionFromDelta methods into single canonical implementation
- Remove debug logger from config-test.yml
- [OPIK-4380] [BE] Fix missing log_comment and centralize search criteria binding
-
Fix SELECT_EXPERIMENT_BY_ID to properly render log_comment metadata
- Use getSTWithLogComment pattern in getExperimentFromAggregates
- Ensures ClickHouse query logging populates workspace/experiment IDs
-
Centralize ExperimentSearchCriteria binding logic
- Create ExperimentSearchCriteriaBinder utility class
- Parameterize filter strategies to support both DAO variants
- Eliminate 29-line duplication between ExperimentDAO and ExperimentAggregatesDAO
- Single source of truth prevents DAOs from getting out of sync
- [OPIK-4380] [BE] Fix createVersionFromDelta consolidation after rebase
-
Update canonical method signature to include new parameters:
- List evaluators
- ExecutionPolicy executionPolicy
- boolean clearExecutionPolicy
-
Update all 5 caller sites to pass new parameters:
- Use changes.evaluators(), changes.executionPolicy(), changes.clearExecutionPolicy() when available
- Pass null/false for auto-generated versions that inherit from base
-
Add imports for EvaluatorItem and ExecutionPolicy
Fixes compilation errors introduced by rebase with upstream changes to DatasetVersionService
- [OPIK-4380] [BE] Address PR review comments - fix type mismatch, extract constants, remove DAO logging
- Fixed BigDecimal[] to Double[] conversion for experiment_scores (matches ClickHouse Float64)
- Extracted FilterStrategy lists to static final constants to avoid repeated allocations
- Added @NonNull validation to populateExperimentAggregate parameter
- Removed DAO layer logging, keeping service-level logging only
- [OPIK-4380] [BE] Extract shared helper for experiment data pagination
Extract streamWithExperimentPagination() helper method to eliminate
duplication in getTracesData(), getSpansData(), and getFeedbackScoresData().All three methods followed identical pattern:
- asyncTemplate.stream with connection
- getSTWithLogComment with cursor flag
- Bind workspace_id, experiment_id, project_id, limit
- Optional cursor binding
- Result mapping
Benefits:
- Single source of truth for pagination binding logic
- Prevents divergence when tweaking cursor/limit bindings
- Reduces code from ~20 lines to ~10 lines per method
- Type-safe generic implementation
Note: CTE redundancy (3x experiment_items scan) is intentional to avoid
passing large trace ID lists as parameters, which would cause performance
issues with 10K+ traces.- [OPIK-4383] [BE] Add Redis stream subscriber for debounced experiment aggregates recomputation
- Add ExperimentDenormalizationConfig implementing StreamConfiguration with debounce, job lock, and per-experiment aggregation lock settings
- Add ExperimentAggregationMessage as stream message record
- Add ExperimentAggregatesSubscriber consuming from the denormalization stream; acquires a workspace-scoped distributed lock per experiment before calling populateAggregations()
- Add experimentDenormalizationEnabled feature flag to ServiceTogglesConfig and FeatureFlags
- Wire ExperimentDenormalizationConfig into OpikConfiguration
- Update config.yml and config-test.yml with full experimentDenormalization block (enabled for tests)
- Add ExperimentAggregatesSubscriberTest covering lifecycle gating and processEvent success/error paths
- Revision 2: Address PR comments - add config defaults, remove toggle, rename tests
- ExperimentDenormalizationConfig: add sensible defaults to all fields so
Dropwizard validation doesn't fail when the config block is absent from
old deployments (config.isEnabled()=false still gates the subscriber) - Remove experimentDenormalizationEnabled service toggle from
ServiceTogglesConfig, FeatureFlags, config.yml and config-test.yml -
the infrastructure gate (config.isEnabled()) is the single control point - Rename lifecycle test methods to camelCase per project conventions:
startSkipsStartupWhenDisabled / stopSkipsShutdownWhenDisabled
-
Revision 3: Add @Max(500) to consumerBatchSize and @NotNull to jobLockWaitTime
-
[OPIK-4380] [BE] Address PR review comments - fix TYPE_REFERENCE visibility, redundant IN subquery, hardcoded context keys, Instant.now in loop, and inline defaultIfNull
-
Revision 4: Address remaining JetoPistola review comments (#7, #8, #10)
- #7: Remove "Used for testing and verification" from getExperimentFromAggregates javadoc
- #8: Replace recursive flatMap with Mono.expand() in populateExperimentItemsInBatches
- #10: Remove unrelated subscribeOn addition from DatasetItemService.createVersionFromDelta
-
Revision 3: Add switchIfEmpty fallback for deleted traces in populateExperimentAggregate
-
Fix tests
-
Revision 6: Move countTotal log from DAO to service layer
Operational logs belong in the service layer, not the DAO.
-
Revision 7: Apply Spotless formatting
-
Revision 8: Make populateAggregations(UUID, int) private
Removes the uncapped public batch size entry point. All callers now go
through the public no-arg overload which reads batchSize safely from config.- [OPIK-4380] [BE] Add evaluation_method support to experiment_aggregates pipeline
- Add ClickHouse migration (000062) to add evaluation_method column to experiment_aggregates table
- Add evaluationMethod field to ExperimentData record
- Update GET_EXPERIMENT_DATA query to read evaluation_method from experiments
- Update INSERT_EXPERIMENT_AGGREGATE to write evaluation_method to experiment_aggregates
- Update SELECT_EXPERIMENT_BY_ID to read evaluation_method from experiment_aggregates
- Fix Experiment record constructor call: insert EvaluationMethod at correct position (10)
- [OPIK-4380] [BE] Extract shared helper for experiment aggregation queries
Reduce copy-paste in getTraceAggregations, getSpanAggregations, and
getFeedbackScoreAggregations by extracting queryExperimentAggregation,
which centralises the context-aware execution, workspace/experiment/project
parameter binding, and singleOrEmpty pattern shared by all three methods.- [OPIK-4380] [BE] Enforce non-null contract on countTotal criteria parameter
Add @NonNull to ExperimentSearchCriteria in the interface and implementation
so that a null argument fails fast with an explicit NullPointerException at
the DAO boundary instead of crashing deep inside buildCountTemplate.- [OPIK-4380] [BE] Fix countTotal ignoring target project IDs in normal path
target_project_ids was only applied inside the project_deleted LEFT JOIN
subquery; the main WHERE had no project restriction, so counts were
workspace-wide. Reuse has_target_projects in the main WHERE so
project_id IN :target_project_ids always takes effect. Also replace
manual null/empty checks with CollectionUtils.isNotEmpty.-
[OPIK-4380] [BE] Apply Spotless formatting
-
[OPIK-4382] [BE] Address PR review comments on experiment aggregates
- Fix :versionId → :version_id parameter naming in SQL templates and bindings
- Fix last_updated_at binding to use item.lastUpdatedAt() instead of Instant.now()
- Fix FEEDBACK_SCORES_AGGREGATED_IS_EMPTY filter: embed generated SQL into templates
instead of hard-coded static condition, and add missing bind calls - Fix RetryUtils log duplication (remove getMessage() + pass exception directly)
- Add batchSize = 1000 default in ExperimentAggregatesConfig
- Extract resolveVersionIdForCriteria helper to deduplicate version-id resolution
- Add null/blank/ClickHouse placeholder guards in extractUuidsFromGroupValues
- Extract loadEntityMap helper to deduplicate getEnrichInfoHolder enrichment logic
- Revision 3: Address PR comments E, F, G, H
- Fix E: Extract shared template/bind helpers in ExperimentAggregatesDAO
- Fix F: Bind experiment_ids as UUID[] instead of String[]
- Fix G+H: Extract getEnrichInfoHolder logic into ExperimentGroupEnricher,
eliminating duplication between ExperimentService and ExperimentAggregatesService
without introducing a direct dependency between them
-
Revision 4: Fix ExperimentServiceTest to include ExperimentGroupEnricher mock
-
[OPIK-4382] [BE] Extract shared Row→ExperimentGroup mappers into ExperimentGroupMappers
Pull the duplicated Row→ExperimentGroupItem and Row→ExperimentGroupAggregationItem
conversion logic from ExperimentDAO and ExperimentAggregatesDAO into a shared
ExperimentGroupMappers utility class. Both DAOs now delegate to the same
toExperimentGroupItem / toExperimentGroupAggregationItem helpers, eliminating
the need to mirror DTO mapping changes in two places.- [OPIK-4382] [BE] Deduplicate bindGroupCriteria into ExperimentGroupMappers
Moves the shared group-criteria binding logic out of ExperimentDAO and
ExperimentAggregatesDAO into ExperimentGroupMappers.bindGroupCriteria(),
following the same pattern as ExperimentSearchCriteriaBinder. Adding or
fixing a criteria binding now only requires a change in one place.- [OPIK-4382] [BE] Extract streamGroupQuery helper and fix null percentiles
-
Deduplicate findGroups/findGroupsAggregations into a single private
streamGroupQuery(queryTemplate, criteria, rowMapper) that differs only
by the query constant and BiFunction row mapper. -
Fix convertToBigDecimal to return null for null/unsupported inputs so
absent p50/p90/p99 entries in getDuration propagate as null rather than
BigDecimal.ZERO, preserving the semantic-null that lets callers apply
COALESCE/fallback logic correctly.
- [OPIK-4382] [BE] Consolidate cost/duration helpers into ExperimentGroupMappers
Promote getCostValue and getDuration to public static in
ExperimentGroupMappers and delete the private copies in ExperimentDAO.
ExperimentDAO.mapToDto now delegates to the shared helpers, so any
future change to cost filtering, duration percentile extraction, or the
BigDecimal conversion only needs to be made in one place.Side-effect: ExperimentDAO.mapToDto also picks up the null-percentile
fix (convertToBigDecimal returns null for absent/unsupported inputs)
that was previously applied only to ExperimentGroupMappers.- [OPIK-4382] [BE] Fix pagination count and add criteria filter tests
- Remove count() OVER () window function from paged query (returned
page-scoped count instead of full result-set count) - Replace with dedicated count query + short-circuit: skip items query
when count == 0, use DatasetItemPage.empty() for that case - Extract DatasetItemResultMapper.buildItemFromRow as public static
helper reused by ExperimentAggregatesDAO - Add parameterized integration tests for ExperimentGroupCriteria
filters (name, types, projectId, combined, empty-result) covering
both findGroups and findGroupsAggregations aggregate paths
- [OPIK-4380] [BE] Extract shared filter helpers into FilterQueryBuilder
Add FilterStrategyParam record, applyFiltersToTemplate and bindFilters
static helpers to FilterQueryBuilder, then replace duplicated per-strategy
loops in DatasetItemVersionDAO and ExperimentAggregatesDAO with single
delegating calls backed by per-DAO strategy constants.- [OPIK-4382] [BE] Consolidate filter helpers in getExperimentItemsStatsFromAggregates
Add EXPERIMENT_ITEMS_STATS_FILTER_STRATEGY_PARAMS and
EXPERIMENT_ITEMS_STATS_BIND_STRATEGIES constants and replace the
per-strategy toAnalyticsDbFilters/bind blocks in
getExperimentItemsStatsFromAggregates with single delegating calls to
FilterQueryBuilder.applyFiltersToTemplate and bindFilters.- Revision 9: Extract shared helpers to eliminate duplication across DAOs
- Create DatasetItemSearchCriteriaMapper: centralizes filters + search flag
wiring for DatasetItemSearchCriteria, shared by DatasetItemVersionDAO and
ExperimentAggregatesDAO - Add ExperimentGroupMappers.applyGroupCriteriaToTemplate: centralizes
ExperimentGroupCriteria → ST template wiring, now shared by ExperimentDAO
and ExperimentAggregatesDAO - Update DatasetItemVersionDAO.addFiltersToTemplate and bindSearchAndFilters
to delegate to DatasetItemSearchCriteriaMapper - Update ExperimentAggregatesDAO.applyDatasetItemFiltersToTemplate and
bindDatasetItemSearchParams to delegate to DatasetItemSearchCriteriaMapper - Update ExperimentDAO.newGroupTemplate and ExperimentAggregatesDAO.newGroupTemplate
to delegate to ExperimentGroupMappers.applyGroupCriteriaToTemplate
- [OPIK-4383] [BE] Fix project_deleted filter and comments_dedup scope in ExperimentAggregatesDAO
- Fix project_deleted filter: use zero UUID sentinel instead of empty string
for FixedString(36) column comparison in FIND_GROUPS and FIND_GROUPS_AGGREGATIONS - Fix comments_dedup CTE: scope trace_id subquery by dataset_id to avoid
scanning the entire workspace's comments table - Add missing streamMaxLen and streamTrimLimit fields to
ExperimentDenormalizationConfig (implements StreamConfiguration interface)
- [OPIK-4383] [BE] Address PR review comments: extract ZERO_UUID constant and fix config comment
- Promote zero UUID sentinel to shared constant in ExperimentGroupMappers
- Use parameterized :zero_uuid binding in SQL templates instead of hardcoded string
- Fix config.yml comment from "Default: 120s" to "Default: 1m"
- [OPIK-4383] [BE] Add streamMaxLen and streamTrimLimit to experimentDenormalization config
下载附件