perf(aggregation): bulk-insert experiment_item_aggregates via ClickHouse HTTP JSONEachRow

The prior INSERT used a StringTemplate that rendered one VALUES tuple per
item with N distinct named parameters (`:id0, :id1, ...`). R2DBC's
named-parameter resolution grew super-linearly with batch size, so
`EXPERIMENT_AGGREGATES_BATCH_SIZE=10000` made the Java-side render + bind
pipeline take ~45s per batch (CH server-side INSERT itself was <200ms).
At the 60s lock TTL this left ~0 headroom: every attempt processed one
10k-item batch, then got cancelled. Smaller batch sizes avoided the
super-linear cost but capped throughput at ~1.5k items/sec.

Replace the R2DBC path for this INSERT with the ClickHouse v2 HTTP
client (`com.clickhouse:client-v2`) POSTing JSONEachRow directly, with
client-request + server-response compression enabled. The rest of the
aggregation pipeline keeps using R2DBC.

At batchSize=10000 the per-batch cost drops from ~45s to ~60-100ms
(bodyMs ~20-40 + executeMs ~35-60). End-to-end a 1M-item experiment
now completes aggregation in ~45s as a single attempt, well within
the 60s lock TTL. ExperimentAggregatesIntegrationTest (126 tests)
passes.

Also wires `EXPERIMENT_AGGREGATES_BATCH_SIZE` through the backend
docker-compose service, defaulting to 10000.

Author: thiagohora

apps/opik-backend/pom.xml

@@ -250,6 +250,11 @@
             <artifactId>clickhouse-http-client</artifactId>
             <version>${clickhouse-java.version}</version>
         </dependency>
+        <dependency>
+            <groupId>com.clickhouse</groupId>
+            <artifactId>client-v2</artifactId>
+            <version>${clickhouse-java.version}</version>
+        </dependency>
         <dependency>
             <groupId>org.apache.httpcomponents.client5</groupId>
             <artifactId>httpclient5</artifactId>

apps/opik-backend/src/main/java/com/comet/opik/domain/experiments/aggregations/ExperimentAggregatesDAO.java

@@ -1,5 +1,10 @@
 package com.comet.opik.domain.experiments.aggregations;
 
+import com.clickhouse.client.api.Client;
+import com.clickhouse.client.api.insert.InsertResponse;
+import com.clickhouse.client.api.insert.InsertSettings;
+import com.clickhouse.client.api.metrics.ServerMetrics;
+import com.clickhouse.data.ClickHouseFormat;
 import com.comet.opik.api.AssertionScoreAverage;
 import com.comet.opik.api.DatasetItem.DatasetItemPage;
 import com.comet.opik.api.EvaluationMethod;
@@ -35,7 +40,6 @@
 import com.comet.opik.infrastructure.auth.RequestContext;
 import com.comet.opik.infrastructure.db.TransactionTemplateAsync;
 import com.comet.opik.utils.JsonUtils;
-import com.comet.opik.utils.template.TemplateUtils;
 import com.fasterxml.jackson.core.type.TypeReference;
 import com.fasterxml.jackson.databind.JsonNode;
 import com.google.inject.ImplementedBy;
@@ -54,8 +58,11 @@
 import org.stringtemplate.v4.ST;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
+import reactor.core.scheduler.Schedulers;
 
+import java.io.ByteArrayInputStream;
 import java.math.BigDecimal;
+import java.nio.charset.StandardCharsets;
 import java.time.Instant;
 import java.util.Arrays;
 import java.util.List;
@@ -83,7 +90,6 @@
 import static com.comet.opik.infrastructure.DatabaseUtils.getSTWithLogComment;
 import static com.comet.opik.utils.AsyncUtils.makeFluxContextAware;
 import static com.comet.opik.utils.AsyncUtils.makeMonoContextAware;
-import static com.comet.opik.utils.template.TemplateUtils.getQueryItemPlaceHolder;
 
 @ImplementedBy(ExperimentAggregatesDAOImpl.class)
 public interface ExperimentAggregatesDAO {
@@ -126,6 +132,7 @@ class ExperimentAggregatesDAOImpl implements ExperimentAggregatesDAO {
     private final @NonNull TransactionTemplateAsync asyncTemplate;
     private final @NonNull FilterQueryBuilder filterQueryBuilder;
     private final @NonNull GroupingQueryBuilder groupingQueryBuilder;
+    private final @NonNull Client clickHouseClient;
 
     /**
      * Filter strategies used for experiment aggregates search binding.
@@ -945,71 +952,6 @@ ORDER BY (workspace_id, project_id, entity_id, id) DESC, last_updated_at DESC
             ;
             """;
 
-    /**
-     * Insert experiment item aggregate
-     */
-    private static final String INSERT_EXPERIMENT_ITEM_AGGREGATE = """
-            INSERT INTO experiment_item_aggregates
-            (
-                workspace_id,
-                id,
-                project_id,
-                experiment_id,
-                dataset_item_id,
-                trace_id,
-                input,
-                output,
-                input_slim,
-                output_slim,
-                metadata,
-                duration,
-                total_estimated_cost,
-                usage,
-                feedback_scores,
-                feedback_scores_array,
-                comments_array_agg,
-                visibility_mode,
-                created_at,
-                last_updated_at,
-                created_by,
-                last_updated_by,
-                execution_policy,
-                assertions_array
-            )
-            SETTINGS log_comment = '<log_comment>'
-            FORMAT Values
-            <items:{item |
-                (
-                    :workspace_id,
-                    :id<item.index>,
-                    :project_id,
-                    :experiment_id<item.index>,
-                    :dataset_item_id<item.index>,
-                    :trace_id<item.index>,
-                    :input<item.index>,
-                    :output<item.index>,
-                    :input_slim<item.index>,
-                    :output_slim<item.index>,
-                    :metadata<item.index>,
-                    :duration<item.index>,
-                    :total_estimated_cost<item.index>,
-                    if(:has_usage<item.index>, mapFromArrays(:usage_keys<item.index>, :usage_values<item.index>), CAST(map() AS Map(String, Int64)) ),
-                    if(:has_feedback_scores<item.index>, mapFromArrays(:feedback_scores_keys<item.index>, CAST(:feedback_scores_values<item.index> AS Array(Decimal64(9)))), CAST(map() AS Map(String, Decimal64(9))) ),
-                    :feedback_scores_array<item.index>,
-                    :comments_array_agg<item.index>,
-                    :visibility_mode<item.index>,
-                    parseDateTime64BestEffort(:created_at<item.index>, 9),
-                    parseDateTime64BestEffort(:last_updated_at<item.index>, 9),
-                    :created_by<item.index>,
-                    :last_updated_by<item.index>,
-                    :execution_policy<item.index>,
-                    :assertions_array<item.index>
-                )
-                <if(item.hasNext)>,<endif>
-            }>
-            ;
-            """;
-
     /**
      * Get experiment items count
      */
@@ -1852,79 +1794,67 @@ private AssertionScoreAggregations createEmptyAssertionScoreAggregations(UUID ex
                 .build();
     }
 
-    private void bindItemsParameters(Statement statement,
-            List<ExperimentItemData> items,
+    private void appendJsonRow(StringBuilder out,
+            String workspaceId,
+            UUID projectId,
+            ExperimentItemData item,
             Map<UUID, TraceData> tracesMap,
             Map<UUID, SpanData> spansMap,
             Map<UUID, FeedbackScoreData> feedbackMap,
             Map<UUID, CommentsData> commentsMap,
             Map<UUID, AssertionData> assertionsMap) {
 
-        for (int i = 0; i < items.size(); i++) {
-            var item = items.get(i);
-
-            TraceData trace = tracesMap.get(item.traceId());
-            SpanData span = spansMap.get(item.traceId());
-            FeedbackScoreData feedback = feedbackMap.get(item.traceId());
-
-            Map<String, Long> usageMap = Optional.ofNullable(span)
-                    .map(SpanData::usage)
-                    .orElse(Map.of());
-            Map<String, BigDecimal> feedbackScoresMap = Optional.ofNullable(feedback)
-                    .map(FeedbackScoreData::feedbackScores)
-                    .orElse(Map.of());
-            String feedbackScoresArray = Optional.ofNullable(feedback)
-                    .map(FeedbackScoreData::feedbackScoresArray)
-                    .orElse(EMPTY_ARRAY_STR);
-
-            statement.bind("id" + i, item.id())
-                    .bind("has_usage" + i, !usageMap.isEmpty())
-                    .bind("has_feedback_scores" + i, !feedbackScoresMap.isEmpty())
-                    .bind("experiment_id" + i, item.experimentId())
-                    .bind("dataset_item_id" + i, item.datasetItemId())
-                    .bind("trace_id" + i, item.traceId())
-                    .bind("input" + i, Optional.ofNullable(trace).map(TraceData::input).orElse(""))
-                    .bind("output" + i, Optional.ofNullable(trace).map(TraceData::output).orElse(""))
-                    .bind("input_slim" + i, Optional.ofNullable(trace).map(TraceData::inputSlim).orElse(""))
-                    .bind("output_slim" + i, Optional.ofNullable(trace).map(TraceData::outputSlim).orElse(""))
-                    .bind("metadata" + i, Optional.ofNullable(trace).map(TraceData::metadata).orElse(""))
-                    .bind("duration" + i, Optional.ofNullable(trace).map(TraceData::duration).orElse(BigDecimal.ZERO))
-                    .bind("total_estimated_cost" + i,
-                            Optional.ofNullable(span).map(SpanData::totalEstimatedCost).orElse(BigDecimal.ZERO))
-                    .bind("feedback_scores_array" + i, feedbackScoresArray)
-                    .bind("comments_array_agg" + i,
-                            Optional.ofNullable(commentsMap.get(item.traceId()))
-                                    .map(CommentsData::commentsArrayAgg)
-                                    .orElse(EMPTY_ARRAY_STR))
-                    .bind("visibility_mode" + i,
-                            Optional.ofNullable(trace)
-                                    .map(TraceData::visibilityMode)
-                                    .map(VisibilityMode::getValue)
-                                    .orElse(VisibilityMode.DEFAULT.getValue()))
-                    .bind("created_at" + i, item.createdAt().toString())
-                    .bind("last_updated_at" + i, item.lastUpdatedAt().toString())
-                    .bind("created_by" + i, item.createdBy())
-                    .bind("last_updated_by" + i, item.lastUpdatedBy())
-                    .bind("execution_policy" + i,
-                            Optional.ofNullable(item.executionPolicy())
-                                    .filter(StringUtils::isNotBlank)
-                                    .orElse(""))
-                    .bind("assertions_array" + i,
-                            Optional.ofNullable(assertionsMap.get(item.traceId()))
-                                    .map(AssertionData::assertionsArray)
-                                    .orElse(EMPTY_ARRAY_STR));
-
-            // Bind array parameters only if maps are not empty
-            var usageArrays = mapToArrays(usageMap, String[]::new, Long[]::new, Long::longValue);
-            statement.bind("usage_keys" + i, usageArrays.keys());
-            statement.bind("usage_values" + i, usageArrays.values());
-
-            var feedbackScoresArrays = mapToArrays(feedbackScoresMap,
-                    String[]::new, BigDecimal[]::new,
-                    v -> BigDecimal.valueOf(v.doubleValue()));
-            statement.bind("feedback_scores_keys" + i, feedbackScoresArrays.keys());
-            statement.bind("feedback_scores_values" + i, feedbackScoresArrays.values());
-        }
+        TraceData trace = tracesMap.get(item.traceId());
+        SpanData span = spansMap.get(item.traceId());
+        FeedbackScoreData feedback = feedbackMap.get(item.traceId());
+
+        Map<String, Long> usageMap = Optional.ofNullable(span).map(SpanData::usage).orElse(Map.of());
+        Map<String, BigDecimal> feedbackScoresMap = Optional.ofNullable(feedback)
+                .map(FeedbackScoreData::feedbackScores).orElse(Map.of());
+
+        var node = JsonUtils.createObjectNode();
+        node.put("workspace_id", workspaceId);
+        node.put("id", item.id().toString());
+        node.put("project_id", projectId.toString());
+        node.put("experiment_id", item.experimentId().toString());
+        node.put("dataset_item_id", item.datasetItemId().toString());
+        node.put("trace_id", item.traceId().toString());
+        node.put("input", Optional.ofNullable(trace).map(TraceData::input).orElse(""));
+        node.put("output", Optional.ofNullable(trace).map(TraceData::output).orElse(""));
+        node.put("input_slim", Optional.ofNullable(trace).map(TraceData::inputSlim).orElse(""));
+        node.put("output_slim", Optional.ofNullable(trace).map(TraceData::outputSlim).orElse(""));
+        node.put("metadata", Optional.ofNullable(trace).map(TraceData::metadata).orElse(""));
+        node.put("duration",
+                Optional.ofNullable(trace).map(TraceData::duration).orElse(BigDecimal.ZERO).doubleValue());
+        node.put("total_estimated_cost",
+                Optional.ofNullable(span).map(SpanData::totalEstimatedCost).orElse(BigDecimal.ZERO)
+                        .toPlainString());
+
+        var usageNode = node.putObject("usage");
+        usageMap.forEach(usageNode::put);
+
+        var feedbackNode = node.putObject("feedback_scores");
+        feedbackScoresMap.forEach((k, v) -> feedbackNode.put(k, v.toPlainString()));
+
+        node.put("feedback_scores_array",
+                Optional.ofNullable(feedback).map(FeedbackScoreData::feedbackScoresArray).orElse(EMPTY_ARRAY_STR));
+        node.put("comments_array_agg",
+                Optional.ofNullable(commentsMap.get(item.traceId())).map(CommentsData::commentsArrayAgg)
+                        .orElse(EMPTY_ARRAY_STR));
+        node.put("visibility_mode",
+                Optional.ofNullable(trace).map(TraceData::visibilityMode).map(VisibilityMode::getValue)
+                        .orElse(VisibilityMode.DEFAULT.getValue()));
+        node.put("created_at", item.createdAt().toString());
+        node.put("last_updated_at", item.lastUpdatedAt().toString());
+        node.put("created_by", item.createdBy());
+        node.put("last_updated_by", item.lastUpdatedBy());
+        node.put("execution_policy",
+                Optional.ofNullable(item.executionPolicy()).filter(StringUtils::isNotBlank).orElse(""));
+        node.put("assertions_array",
+                Optional.ofNullable(assertionsMap.get(item.traceId())).map(AssertionData::assertionsArray)
+                        .orElse(EMPTY_ARRAY_STR));
+
+        out.append(node).append('\n');
     }
 
     private Mono<Long> insertExperimentItemAggregates(
@@ -1948,27 +1878,41 @@ private Mono<Long> insertExperimentItemAggregates(
         Map<UUID, AssertionData> assertionsMap = assertionsData.stream()
                 .collect(Collectors.toMap(AssertionData::traceId, Function.identity()));
 
-        return asyncTemplate.nonTransaction(connection -> {
-            return makeMonoContextAware((userName, workspaceId) -> {
-
-                List<TemplateUtils.QueryItem> queryItems = getQueryItemPlaceHolder(items.size());
-
-                var template = getSTWithLogComment(INSERT_EXPERIMENT_ITEM_AGGREGATE,
-                        "insertExperimentItemAggregate", workspaceId, userName, items.size())
-                        .add("items", queryItems);
-
-                var statement = connection.createStatement(template.render())
-                        .bind("workspace_id", workspaceId)
-                        .bind("project_id", projectId);
+        return insertExperimentItems(projectId, items, tracesMap, spansMap, feedbackMap, commentsMap, assertionsMap);
+    }
 
-                // Bind item parameters in batch
-                bindItemsParameters(statement, items, tracesMap, spansMap, feedbackMap, commentsMap, assertionsMap);
+    private Mono<Long> insertExperimentItems(UUID projectId,
+            List<ExperimentItemData> items,
+            Map<UUID, TraceData> tracesMap,
+            Map<UUID, SpanData> spansMap,
+            Map<UUID, FeedbackScoreData> feedbackMap,
+            Map<UUID, CommentsData> commentsMap,
+            Map<UUID, AssertionData> assertionsMap) {
 
-                return Mono.from(statement.execute())
-                        .flatMapMany(Result::getRowsUpdated)
-                        .reduce(0L, Long::sum);
-            });
-        });
+        return makeMonoContextAware((userName, workspaceId) -> Mono.fromCallable(() -> {
+            StringBuilder body = new StringBuilder();
+            items.forEach(item -> appendJsonRow(body, workspaceId, projectId, item,
+                    tracesMap, spansMap, feedbackMap, commentsMap, assertionsMap));
+            byte[] payload = body.toString().getBytes(StandardCharsets.UTF_8);
+
+            String logComment = "insert_experiment_item_aggregate:%s:%s:%d".formatted(
+                    workspaceId, userName == null ? "" : userName, items.size());
+
+            var settings = new InsertSettings()
+                    .logComment(logComment)
+                    .serverSetting("date_time_input_format", "best_effort");
+
+            try (InsertResponse response = clickHouseClient.insert(
+                    "experiment_item_aggregates",
+                    new ByteArrayInputStream(payload),
+                    ClickHouseFormat.JSONEachRow,
+                    settings).get()) {
+                return response.getMetrics().getMetric(ServerMetrics.NUM_ROWS_WRITTEN).getLong();
+            }
+        }).subscribeOn(Schedulers.boundedElastic())
+                .doOnError(err -> log.error(
+                        "Failed to insert experiment item aggregates: items='{}'",
+                        items.size(), err)));
     }
 
     // Row mapping methods

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

@@ -1,5 +1,6 @@
 package com.comet.opik.infrastructure.db;
 
+import com.clickhouse.client.api.Client;
 import com.comet.opik.infrastructure.ClickHouseLogAppenderConfig;
 import com.comet.opik.infrastructure.DatabaseAnalyticsFactory;
 import com.comet.opik.infrastructure.OpikConfiguration;
@@ -38,6 +39,27 @@ public ConnectionFactory getConnectionFactory() {
         return connectionFactory;
     }
 
+    @Provides
+    @Singleton
+    public DatabaseAnalyticsFactory getDatabaseAnalyticsFactory() {
+        return databaseAnalyticsFactory;
+    }
+
+    @Provides
+    @Singleton
+    public Client getClickHouseClient() {
+        var config = databaseAnalyticsFactory;
+        String endpoint = "%s://%s:%d/".formatted(config.getProtocol().getValue(), config.getHost(), config.getPort());
+        return new Client.Builder()
+                .addEndpoint(endpoint)
+                .setUsername(config.getUsername())
+                .setPassword(config.getPassword())
+                .setDefaultDatabase(config.getDatabaseName())
+                .compressClientRequest(true)
+                .compressServerResponse(true)
+                .build();
+    }
+
     @Provides
     @Singleton
     @Named("Database Analytics Database Name")