Enhance batch operations configuration by adding analytics settings for bulk update chunk size and stream buffer size. Update DAO implementations to utilize these new configuration parameters for improved performance in bulk operations.

Author: YarivHashaiComet

apps/opik-backend/config.yml

@@ -204,6 +204,13 @@ batchOperations:
     # Description: The maximal number of ids to be used for IN clause. Find requests with a larger number of ids will
     # involve the use of temp tables for querying
     maxExperimentInClauseSize: ${BATCH_OPERATIONS_MAX_EXPERIMENT_IN_CLAUSE_SIZE:-5000}
+  analytics:
+    # Default: 100
+    # Description: Chunk size for bulk update operations (traces, spans, threads). ClickHouse performs best with batch sizes of 100-1000 records per query.
+    bulkUpdateChunkSize: ${BATCH_OPERATIONS_BULK_UPDATE_CHUNK_SIZE:-100}
+    # Default: 1000
+    # Description: Buffer size for streaming operations to batch records for efficient processing
+    streamBufferSize: ${BATCH_OPERATIONS_STREAM_BUFFER_SIZE:-1000}
 
 # Configuration for rate limit. This is not enabled by default for open source installations.
 # If enabled, rate limit is applied to creation and update of various entities including traces, spans, projects,

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

@@ -2365,12 +2365,11 @@ public Mono<Void> bulkUpdateTags(@NonNull Map<UUID, SpanUpdate> idToUpdateMap) {
                 .map(id -> idToUpdateMap.get(id).tags().toArray(new String[0]))
                 .toList();
 
-        int CHUNK_SIZE = 100;
-
-        return Flux.range(0, (ids.size() + CHUNK_SIZE - 1) / CHUNK_SIZE)
+        int chunkSize = configuration.getBatchOperations().getAnalytics().getBulkUpdateChunkSize();
+        return Flux.range(0, (ids.size() + chunkSize - 1) / chunkSize)
                 .flatMap(chunkIndex -> {
-                    int start = chunkIndex * CHUNK_SIZE;
-                    int end = Math.min(start + CHUNK_SIZE, ids.size());
+                    int start = chunkIndex * chunkSize;
+                    int end = Math.min(start + chunkSize, ids.size());
 
                     List<UUID> chunkIds = ids.subList(start, end);
                     List<String[]> chunkTags = tagsArrays.subList(start, end);

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

@@ -3614,12 +3614,11 @@ public Mono<Void> bulkUpdateTags(@NonNull Map<UUID, TraceUpdate> idToUpdateMap)
                 .map(id -> idToUpdateMap.get(id).tags().toArray(new String[0]))
                 .toList();
 
-        int CHUNK_SIZE = 100;
-
-        return Flux.range(0, (ids.size() + CHUNK_SIZE - 1) / CHUNK_SIZE)
+        int chunkSize = configuration.getBatchOperations().getAnalytics().getBulkUpdateChunkSize();
+        return Flux.range(0, (ids.size() + chunkSize - 1) / chunkSize)
                 .flatMap(chunkIndex -> {
-                    int start = chunkIndex * CHUNK_SIZE;
-                    int end = Math.min(start + CHUNK_SIZE, ids.size());
+                    int start = chunkIndex * chunkSize;
+                    int end = Math.min(start + chunkSize, ids.size());
 
                     List<UUID> chunkIds = ids.subList(start, end);
                     List<String[]> chunkTags = tagsArrays.subList(start, end);

apps/opik-backend/src/main/java/com/comet/opik/domain/threads/TraceThreadDAO.java

@@ -4,6 +4,7 @@
 import com.comet.opik.api.TraceThreadStatus;
 import com.comet.opik.api.TraceThreadUpdate;
 import com.comet.opik.api.events.ProjectWithPendingClosureTraceThreads;
+import com.comet.opik.infrastructure.OpikConfiguration;
 import com.comet.opik.infrastructure.db.TransactionTemplateAsync;
 import com.comet.opik.infrastructure.instrumentation.InstrumentAsyncUtils;
 import com.comet.opik.utils.TemplateUtils;
@@ -284,6 +285,7 @@ AND last_updated_at < parseDateTime64BestEffort(:last_updated_at, 6)
 
     private final @NonNull TransactionTemplateAsync asyncTemplate;
     private final @NonNull ConnectionFactory connectionFactory;
+    private final @NonNull OpikConfiguration configuration;
 
     @Override
     public Mono<Long> save(@NonNull List<TraceThreadModel> traceThreads) {
@@ -569,7 +571,7 @@ public Flux<List<TraceThreadModel>> streamPendingClosureThreads(@NonNull UUID pr
 
             return makeFluxContextAware(bindWorkspaceIdToFlux(statement))
                     .flatMap(result -> result.map((row, rowMetadata) -> TraceThreadMapper.INSTANCE.mapFromRow(row)));
-        }).buffer(1000);
+        }).buffer(configuration.getBatchOperations().getAnalytics().getStreamBufferSize());
     }
 
     private void bindTemplateParam(TraceThreadCriteria criteria, ST template) {
@@ -620,7 +622,10 @@ public Flux<ThreadIdWithTagsAndMetadata> getIdsTagsAndMetadata(@NonNull List<UUI
                     var statement = connection.createStatement(SELECT_IDS_TAGS_AND_METADATA)
                             .bind("ids", ids);
 
-                    return makeFluxContextAware(bindWorkspaceIdToFlux(statement));
+                    Segment segment = startSegment("trace_threads", "Clickhouse", "get_ids_tags_and_metadata");
+
+                    return makeFluxContextAware(bindWorkspaceIdToFlux(statement))
+                            .doFinally(signalType -> endSegment(segment));
                 })
                 .flatMap(result -> result.map((row, metadata) -> {
                     var tags = row.get("tags", String[].class);
@@ -679,12 +684,11 @@ public Mono<Void> bulkUpdateTags(@NonNull Map<ThreadIdWithTagsAndMetadata, Trace
                 .toList();
         UUID projectId = metas.get(0).projectId(); // All should have same projectId
 
-        int CHUNK_SIZE = 100;
-
-        return Flux.range(0, (ids.size() + CHUNK_SIZE - 1) / CHUNK_SIZE)
+        int chunkSize = configuration.getBatchOperations().getAnalytics().getBulkUpdateChunkSize();
+        return Flux.range(0, (ids.size() + chunkSize - 1) / chunkSize)
                 .flatMap(chunkIndex -> {
-                    int start = chunkIndex * CHUNK_SIZE;
-                    int end = Math.min(start + CHUNK_SIZE, ids.size());
+                    int start = chunkIndex * chunkSize;
+                    int end = Math.min(start + chunkSize, ids.size());
 
                     List<UUID> chunkIds = ids.subList(start, end);
                     List<String[]> chunkTags = tagsArrays.subList(start, end);

apps/opik-backend/src/main/java/com/comet/opik/infrastructure/BatchOperationsConfig.java

@@ -14,7 +14,26 @@ public static class DatasetsConfig {
         private @Valid @JsonProperty @Positive int maxExperimentInClauseSize;
     }
 
+    @Data
+    public static class AnalyticsConfig {
+        /**
+         * Chunk size for bulk update operations (traces, spans, threads).
+         * ClickHouse performs best with batch sizes of 100-1000 records per query.
+         * Default: 100
+         */
+        private @Valid @JsonProperty @Positive int bulkUpdateChunkSize = 100;
+
+        /**
+         * Buffer size for streaming operations to batch records for efficient processing.
+         * Default: 1000
+         */
+        private @Valid @JsonProperty @Positive int streamBufferSize = 1000;
+    }
+
     @Valid @JsonProperty
     @NotNull private DatasetsConfig datasets;
 
+    @Valid @JsonProperty
+    @NotNull private AnalyticsConfig analytics = new AnalyticsConfig();
+
 }