github-featured / comet-ml--opik

发布

0 发布 8,340 Git 标签

RSS 订阅源
  • 1.10.40-4517 5204da28f1
    比较

    [OPIK-4387] [BE] feat: wire aggregation publisher into finishExperiments endpoint (#5583)

    frostbyte_neo 发布于 2026-03-16 11:39:50 +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] Add experiment aggregate event listener and no-op publisher

    • Revision 2: Fix missing import for ExperimentAggregationPublisher

    • [OPIK-4383] [BE] Add ExperimentAggregationPublisher, ExperimentDenormalizationJob and tests

    • ExperimentAggregationPublisher: debounces experiment aggregation triggers
      by writing compound workspaceId:experimentId members to a Redis ZSET scored
      by expiry timestamp (now + debounceDelay), plus a hash storing the userName
      with TTL=2×debounceDelay to handle stale entries.
    • ExperimentDenormalizationJob: @Every("5s") job that reads ZSET members with
      score <= now, publishes ExperimentAggregationMessage to the Redis stream,
      then cleans up the ZSET entry and hash bucket. Handles stale entries
      (expired hash) by removing the orphaned ZSET member without publishing.
    • Fix processExperiment reactive chain: avoided double index.remove by
      returning Mono from flatMap branches so switchIfEmpty is only
      triggered when the bucket is truly empty.
    • ExperimentAggregationPublisherTest: integration tests with real Redis
      container verifying ZSET membership, score, userName storage, TTL,
      workspace isolation, and debounce deduplication.
    • ExperimentDenormalizationJobTest: unit tests with Mockito covering disabled
      config, lock not acquired, empty ZSET, happy path, stale entry, and batch.

    • Fix tests setup

    • [OPIK-4383] [BE] Address PR review: move DAO logs to service layer

    • [OPIK-4383] [BE] Address PR review: extract shared DAO helper and fix log placement

    • [OPIK-4383] [BE] Short-circuit deleteByTraceIds when no spans found

    Skip delete, cascading operations, and SpansDeleted event when
    getSpanIdsForTraces returns an empty set, preserving the original
    no-op behaviour and avoiding the Preconditions.checkArgument failure
    in SpanDAO.deleteByIds.

    • [OPIK-4383] [BE] Fix cascade deletion failures after trace delete

    Two bugs prevented spans and attachments from being deleted when a trace
    was deleted via the event-driven cascade:

    1. FeedbackScoreService.deleteByTraceIds/deleteBySpanIds had @NonNull on
      projectId which threw NPE when TracesDeleted.projectId() was null.
      EventInterceptor swallowed the NPE, stopping the entire cascade chain.
      Fix: remove @NonNull since the DAO already handles null safely via
      Optional.ofNullable(projectId).

    2. SpanDAO.DELETE_BY_IDS had the wrong column (trace_id) and parameter
      name (span_ids) — the ClickHouse R2DBC driver could not resolve :span_ids
      as a named parameter in the DELETE statement. Fixed by using id IN :ids
      to match the working pattern in TraceDAO.DELETE_BY_ID.

    • [OPIK-4383] [BE] Address PR review comments on ExperimentDenormalizationJob
    • Centralize Redis constants (EXPERIMENT_KEY_PREFIX, USER_NAME_FIELD,
      MEMBER_SEPARATOR) in ExperimentDenormalizationConfig
    • Change ExperimentAggregationPublisher.publish() to return Mono
      instead of void, so errors propagate to callers
    • Make job interval configurable via jobs map in config.yml
    • Fix onErrorContinue logging: remove getMessage() duplication
    • Demote per-experiment logs from INFO to DEBUG
    • Add ZSET pagination using expand() to avoid materializing entire range
    • Update tests for all changes
    • Fix @Every job interval config key casing and add jobs section to test config

    The dropwizard-jobs framework uses WordUtils.uncapitalize(class.getSimpleName())
    to look up the interval in the jobs map, so the key must be
    'experimentDenormalizationJob' (lowercase first letter). Also adds the missing
    jobs section and jobBatchSize to config-test.yml.

    • Replace @Every annotation with programmatic Quartz scheduling

    Remove @Every from ExperimentDenormalizationJob and schedule it
    programmatically in OpikGuiceyLifecycleEventListener, following the
    same pattern as TraceThreadsClosingJob. Add jobInterval config field
    to ExperimentDenormalizationConfig. Remove the jobs YAML section that
    caused deserialization errors with JobConfiguration's immutable map.

    • Add experiment context to error log and extract publishIfNotEmpty helper
    • Include experimentId and workspaceId in onExperimentUpdated error log
    • Extract publishIfNotEmpty helper to deduplicate filter+publish logic
      across triggerByExperimentIds, triggerByTraceIds, triggerBySpanIds
    • Fix NPE in ExperimentAggregateEventListenerTest mock setup

    Stub publisher.publish() to return Mono.empty() in setUp so
    .subscribe() calls in production code don't NPE on null.

    • [OPIK-4385] [BE] Use pre-computed aggregation tables for experiment endpoints

    Apply UNION ALL hybrid pattern to ExperimentDAO (FIND, FIND_GROUPS,
    FIND_GROUPS_AGGREGATIONS) and ExperimentItemDAO (STREAM,
    SELECT_DATASET_ITEM_VERSIONS_WITH_EXPERIMENT_ITEMS_COUNT,
    SELECT_DATASET_ITEM_VERSIONS_WITH_EXPERIMENT_ITEMS,
    SELECT_DATASET_ITEM_VERSIONS_WITH_EXPERIMENT_ITEMS_STATS) so that
    experiments present in experiment_aggregates / experiment_item_aggregates
    use pre-computed values, while others fall back to live JOIN computation.

    Add ExperimentAggregatesIntegrationTest covering all 7 affected queries
    with parameterized filter, pagination, and consistency scenarios.

    • [OPIK-4386] [BE] Trigger lazy aggregation via publisher on GET experiment by ID

    When fetching an experiment by ID, if the experiment is in COMPLETED or
    CANCELLED state and is not yet present in the experiment_aggregates table,
    enqueue it for aggregation using ExperimentAggregationPublisher instead of
    computing aggregations synchronously. The check and publish are performed
    off the critical path via doOnEach, so the caller receives the experiment
    immediately without waiting for the side effect to complete.

    • [OPIK-4384] [BE] Fix missing zero_uuid binding and experiment_scores sort alias
    • Bind zero_uuid parameter in getById, getByIds, and get(ExperimentStreamRequest)
      methods that use the FIND query; the UNION ALL refactor introduced an
      experiments_from_aggregates CTE that requires this parameter but only the
      main find() method was binding it, causing 500 errors on those paths
    • Fix SortingQueryBuilder to reference the outer column alias experiment_scores_agg
      instead of es.experiment_scores; the ORDER BY sits outside the UNION ALL so the
      inner es alias is out of scope, while experiment_scores_agg is the consistent
      output alias exposed by both branches
    • [OPIK-4384] [BE] Fix null row injection from LEFT JOIN miss in feedback_scores and comments aggregation

    Pre-aggregate feedback_scores_final and comments_final into subqueries
    (GROUP BY entity_id) before LEFT JOIN in DatasetItemVersionDAO.STREAM.
    When a LEFT JOIN has no match against a pre-aggregated subquery the
    joined columns are NULL, so any(NULL) returns NULL instead of a
    default-valued row with epoch timestamps that caused Instant.parse()
    failures.

    Also adds a regression test covering the no-scores path in
    ExperimentAggregatesIntegrationTest.

    • [OPIK-4383] [BE] Remove DAO-level log.info from ExperimentAggregatesDAO methods

    Move operational logging responsibility to the service layer, consistent
    with earlier fixes for ExperimentItemDAO and SpanDAO in this PR.

    • Remove accidentally committed doc files

    These files were introduced during merge resolution but should
    not be part of the branch.

    • [OPIK-4383] [BE] refactor: extract triggerAggregation helper to centralize guard+publish flow

    • [OPIK-4386] [BE] fix: demote lazy aggregation check log to DEBUG

    • [OPIK-4387] [BE] feat: wire aggregation publisher into finishExperiments endpoint

    Chain experimentAggregationPublisher.publish() after AlertEvent in
    finishExperiments() so experiments finished via POST /v1/private/experiments/finish
    are published to Redis for aggregation computation.

    • [OPIK-4383] [BE] fix: restore TagOperations.tagUpdateFragment in SpanDAO BULK_UPDATE

    Restores proper tag handling in SpanDAO.BULK_UPDATE query that was
    regressed to a simple arrayConcat. Now uses TagOperations.tagUpdateFragment()
    which provides arrayDistinct(), tag limit enforcement (max 50), and
    tags_to_add/tags_to_remove support. Also adds the required
    short_circuit_function_evaluation SETTINGS for throwIf evaluation.

    • [OPIK-4387] [BE] feat: add stream trimming to experiment denormalization XADD

    Add streamMaxLen and streamTrimLimit configuration to bound Redis stream
    growth on the experiment denormalization producer (ExperimentDenormalizationJob).
    Uses Redisson's trimNonStrict().maxLen().limit() API for approximate trimming.

    • [OPIK-4387] [BE] fix: make aggregation publish best-effort in finishExperiments

    Swallow and log Redis/publish errors so finishExperiments returns 204
    even when Redis is down. Aggregation will be retried by the lazy trigger
    or next job cycle.

    • [OPIK-4387] [BE] refactor: centralize Redis stream XADD trimming in RedisStreamUtils

    Extract duplicate StreamAddArgs.entry().trimNonStrict().maxLen().limit()
    into RedisStreamUtils.buildAddArgs() so stream trimming settings live in
    one place. Updates all 5 producers.

    • [OPIK-4387] [BE] fix: defer aggregation publish and update test for best-effort behavior

    Wrap aggregation publisher in Mono.defer() so it subscribes only after
    upstream completes, and update unit test to expect completion instead of
    error propagation.

    • Adding InterruptableJob

    • [OPIK-4383] [BE] Address PR review: expand safety valve, env var prefix

    • Add batchSize-capped iteration counter to expand() to prevent
      infinite loops when ZSET entries fail to be removed
    • Rename EXPERIMENT_DENORM_JOB_INTERVAL to OPIK_EXPERIMENT_DENORM_JOB_INTERVAL
      to follow the OPIK_ prefix convention
    • [OPIK-4384] [BE] Add branch optimization and CTE split to experiment queries

    Use pre-computed experiment_aggregates table to optimize query execution:

    • Add has_aggregated/has_raw flags to skip unnecessary UNION ALL branches in FIND/FIND_COUNT
    • Add getAggregationBranchCounts pre-query to determine which branches are needed
    • Apply CTE split pattern to FIND_GROUPS and FIND_GROUPS_AGGREGATIONS
    • Update getById to leverage branch optimization via single-ID branch count query
    • Add <if(id)> filter to SELECT_AGGREGATED_EXPERIMENT_IDS for getById support
    • [OPIK-4384] [BE] Add missing 7-arg overload for getDatasetItemsWithExperimentItems

    Fix test compilation error from merge: the remote branch added callers
    with (UUID, List, null, null, List, String, String) signature
    which needs a bridge overload to the 9-arg method.

    • [OPIK-4384] [BE] Add conditional LIMIT push-up, missing CTE, and fix test precision
    • Add conditional LIMIT push-up in STREAM query: push LIMIT into CTE
      when only one branch (raw or aggregated) is active for performance
    • Add missing experiment_item_aggr_trace_scope CTE for aggregated branch
    • Add AggregatedExperimentCounts record for experiment-level branching
    • Fix MultiValueFeedbackScoresE2ETest precision assertion: use isEqualTo
      instead of isEqualByComparingTo to respect custom BigDecimal comparator

    • [OPIK-4384] [BE] Push OFFSET into top_dataset_items CTE and fix BigDecimal comparator in DatasetsResourceTest

    • [OPIK-4384] [BE] Add pass rate aggregation to experiment aggregates

    Add pass_rate, passed_count, and total_count columns to experiment_aggregates
    table and compute them during aggregation. Update ExperimentDAO queries to
    select these columns from both raw and aggregated paths, returning NULL for
    non-evaluation-suite experiments.

    • Fix format

    • Fix get by id

    • Fix mapping

    • Fix mapping

    • [OPIK-4384] [BE] Use pre-aggregated comments from aggregate tables with ISO 8601 date formatting

    Update retrieval queries in ExperimentDAO, DatasetItemVersionDAO, and ExperimentAggregatesDAO
    to read comments_array_agg as JSON String from aggregate tables instead of live-querying the
    comments table. Ensure UNION ALL type compatibility by wrapping raw paths with toJSONString()
    and formatting dates as ISO 8601 for proper Jackson deserialization.

    • [OPIK-4386] [BE] Increase debounceDelay in test config to prevent race condition

    The denormalization job was processing finished experiments during test
    execution with incomplete ClickHouse data, causing stale aggregated
    values to be returned instead of fresh raw computations.

    • [OPIK-4384] [BE] Use parameterized binding for dynamic sort keys and add deterministic tiebreaker
    • Replace literal string interpolation in getTopSortExpression with
      parameterized bind variables (sf.bindKey()) to prevent SQL injection
    • Remove fieldMapping filter from bindDynamicKeys so all dynamic keys
      are bound, including those used in the top_sorting SELECT expression
    • Add deterministic tiebreaker (id DESC / dataset_item_id DESC) to both
      the push-top-limit CTE and the main ORDER BY for consistent pagination
    • Fix experiment_items deduplication: use FINAL where DISTINCT was used
      and vice versa for consistency across query branches
    • [OPIK-4384] [BE] Add mixed-state aggregation test for UNION ALL hybrid

    Test creates 3 experiments, aggregates only 1, and queries all 3 to
    exercise the UNION ALL hybrid path where has_aggregated and has_raw
    are both true simultaneously.

    • [OPIK-4384] [BE] Add isNotEmpty assertions to parameterized filter tests

    Ensure filter scenarios actually match data by asserting content()
    is not empty before and after aggregation in all parameterized filter
    tests (find, findGroups, findGroupsAggregations).

    • [OPIK-4384] [BE] refactor: extract assertion helpers to remove duplication in ExperimentAggregatesIntegrationTest

    • [OPIK-4384] [BE] refactor: rename parseFlexibleInstant to parseInstant in FeedbackScoreMapper

    • [OPIK-4384] [BE] Make LIMIT unconditional in FIND query

    The LIMIT clause was gated on filter/sort flags, so plain paged requests
    (only limit/offset) at the outer query level would not emit LIMIT.
    Simplify to always emit LIMIT when the limit parameter is provided.

    • [OPIK-4384] [BE] Fix comment ordering assertion in tests

    ClickHouse groupUniqArray does not guarantee ordering, so comment
    assertions must use ignoringCollectionOrder to avoid flaky failures.

    • [OPIK-4384] [BE] Add branch conditionals to FIND_GROUPS/FIND_GROUPS_AGGREGATIONS and revert unconditional LIMIT
    • Wrap SELECT branches in FIND_GROUPS and FIND_GROUPS_AGGREGATIONS with
      <if(has_aggregated)>/<if(has_raw)> conditionals to skip unnecessary
      branches when all experiments are aggregated or all are raw
    • Add no-args getAggregationBranchCounts() overload for workspace-only
      pre-query (used by group/aggregation queries that lack experiment IDs)
    • Update executeQueryWithTargetProjects to run both pre-queries in
      parallel via Mono.zip
    • Revert commit 215a3f96a7 (unconditional LIMIT) which caused double
      LIMIT/OFFSET bug: CTE-level LIMIT + outer LIMIT made page 2+ return
      0 results. The complex conditional is correct — outer LIMIT is only
      needed when post-CTE processing may alter the result set.
    • [OPIK-4384] [BE] Add branch conditionals to SELECT_DATASET_ITEM_VERSIONS_WITH_EXPERIMENT_ITEMS_COUNT

    Wrap the UNION ALL in the count query with <if(has_aggregated)>/<if(has_raw)>
    conditionals to skip unnecessary branches. Pass branch flags through
    getCountWithExperimentFilters from the existing pre-query results.

    • [OPIK-4384] [BE] Fix ClickHouse column resolution in COUNT query

    Alias dataset_item_id as di_id in the COUNT subquery branches
    to avoid column name ambiguity when ClickHouse 25.3's query
    analyzer resolves COUNT(DISTINCT dataset_item_id) through a
    LEFT JOIN with dataset_items_resolved which also has that column.

    • [OPIK-4384] [BE] Use pre-computed comments in STREAM query and fix UNION ALL type mismatch

    Aggregated branch now reads comments_array_agg directly from experiment_item_aggregates
    instead of doing an expensive JOIN to the comments table. Raw branch converts comments
    to JSON String via toJSONString(CAST(...)) so both branches output compatible types.

    • [OPIK-4384] [BE] Fix target_project_ids bind error in FIND_GROUPS aggregated branch

    • Fix issues

    • [OPIK-4387] [BE] Fix missing closing brace in ExperimentServiceTest

    下载附件