OPIK-2164: Add more async inserts

thiagohora · thiagohora · commit 83ce8dafc538 · 2025-08-01T16:58:05.000+02:00
diff --git a/apps/opik-backend/src/main/java/com/comet/opik/domain/FeedbackScoreDAO.java b/apps/opik-backend/src/main/java/com/comet/opik/domain/FeedbackScoreDAO.java
@@ -3,7 +3,9 @@
 import com.comet.opik.api.FeedbackScore;
 import com.comet.opik.api.FeedbackScoreItem;
 import com.comet.opik.api.ScoreSource;
+import com.comet.opik.infrastructure.OpikConfiguration;
 import com.comet.opik.infrastructure.db.TransactionTemplateAsync;
+import com.comet.opik.utils.ClickhouseUtils;
 import com.comet.opik.utils.TemplateUtils;
 import com.google.common.base.Preconditions;
 import com.google.inject.ImplementedBy;
@@ -92,7 +94,7 @@ INSERT INTO feedback_scores(
                 source,
                 created_by,
                 last_updated_by
-            )
+            ) <settings_clause>
             VALUES
                 <items:{item |
                     (
@@ -253,6 +255,7 @@ ORDER BY (workspace_id, project_id, entity_type, entity_id, name) DESC, last_upd
             """;
 
     private final @NonNull TransactionTemplateAsync asyncTemplate;
+    private final @NonNull OpikConfiguration opikConfiguration;
 
     @Override
     @WithSpan
@@ -338,6 +341,10 @@ public Mono<Long> scoreBatchOf(@NonNull EntityType entityType,
 
             ST template = TemplateUtils.getBatchSql(BULK_INSERT_FEEDBACK_SCORE, scores.size());
 
+            if (opikConfiguration.getAsyncInsert().enabled()) {
+                template.add("settings_clause", ClickhouseUtils.ASYNC_INSERT);
+            }
+
             var statement = connection.createStatement(template.render());
 
             bindParameters(entityType, scores, statement);
diff --git a/apps/opik-backend/src/main/java/com/comet/opik/domain/evaluators/AutomationRuleEvaluatorLogsDAO.java b/apps/opik-backend/src/main/java/com/comet/opik/domain/evaluators/AutomationRuleEvaluatorLogsDAO.java
@@ -3,6 +3,8 @@
 import ch.qos.logback.classic.spi.ILoggingEvent;
 import com.comet.opik.api.LogCriteria;
 import com.comet.opik.api.LogItem;
+import com.comet.opik.infrastructure.OpikConfiguration;
+import com.comet.opik.utils.ClickhouseUtils;
 import com.google.inject.ImplementedBy;
 import io.r2dbc.spi.ConnectionFactory;
 import io.r2dbc.spi.Row;
@@ -33,8 +35,8 @@ public interface AutomationRuleEvaluatorLogsDAO extends UserLogTableDAO {
 
     List<String> CUSTOM_MARKER_KEYS = List.of("trace_id", "thread_model_id");
 
-    static AutomationRuleEvaluatorLogsDAO create(ConnectionFactory factory) {
-        return new AutomationRuleEvaluatorLogsDAOImpl(factory);
+    static AutomationRuleEvaluatorLogsDAO create(ConnectionFactory factory, OpikConfiguration configuration) {
+        return new AutomationRuleEvaluatorLogsDAOImpl(factory, configuration);
     }
 
     Mono<LogPage> findLogs(LogCriteria criteria);
@@ -48,6 +50,7 @@ class AutomationRuleEvaluatorLogsDAOImpl implements AutomationRuleEvaluatorLogsD
 
     private static final String INSERT_STATEMENT = """
             INSERT INTO automation_rule_evaluator_logs (timestamp, level, workspace_id, rule_id, message, markers)
+            <settings_clause>
             VALUES <items:{item |
                 (
                     parseDateTime64BestEffort(:timestamp<item.index>, 9),
@@ -72,6 +75,7 @@ INSERT INTO automation_rule_evaluator_logs (timestamp, level, workspace_id, rule
             """;
 
     private final @NonNull ConnectionFactory connectionFactory;
+    private final @NonNull OpikConfiguration opikConfiguration;
 
     public Mono<LogPage> findLogs(@NonNull LogCriteria criteria) {
         return Mono.from(connectionFactory.create())
@@ -135,6 +139,10 @@ public Mono<Void> saveAll(@NonNull List<ILoggingEvent> events) {
                 .flatMapMany(connection -> {
                     var template = new ST(INSERT_STATEMENT);
 
+                    if (opikConfiguration.getAsyncInsert().enabled()) {
+                        template.add("settings_clause", ClickhouseUtils.ASYNC_INSERT);
+                    }
+
                     List<QueryItem> queryItems = getQueryItemPlaceHolder(events.size());
                     template.add("items", queryItems);
                     Statement statement = connection.createStatement(template.render());
diff --git a/apps/opik-backend/src/main/java/com/comet/opik/infrastructure/db/DatabaseAnalyticsModule.java b/apps/opik-backend/src/main/java/com/comet/opik/infrastructure/db/DatabaseAnalyticsModule.java
@@ -28,7 +28,7 @@ protected void configure() {
 
         // Initialize the UserFacingRuleLollingFactory
         UserFacingLoggingFactory.init(connectionFactory, clickHouseLogAppenderConfig.getBatchSize(),
-                clickHouseLogAppenderConfig.getFlushIntervalDuration());
+                clickHouseLogAppenderConfig.getFlushIntervalDuration(), configuration(OpikConfiguration.class));
     }
 
     @Provides
diff --git a/apps/opik-backend/src/main/java/com/comet/opik/infrastructure/db/IdGeneratorImpl.java b/apps/opik-backend/src/main/java/com/comet/opik/infrastructure/db/IdGeneratorImpl.java
@@ -1,5 +1,6 @@
-package com.comet.opik.domain;
+package com.comet.opik.infrastructure.db;
 
+import com.comet.opik.domain.IdGenerator;
 import com.fasterxml.uuid.Generators;
 import com.fasterxml.uuid.impl.TimeBasedEpochGenerator;
 
diff --git a/apps/opik-backend/src/main/java/com/comet/opik/infrastructure/db/IdGeneratorModule.java b/apps/opik-backend/src/main/java/com/comet/opik/infrastructure/db/IdGeneratorModule.java
@@ -1,7 +1,6 @@
 package com.comet.opik.infrastructure.db;
 
 import com.comet.opik.domain.IdGenerator;
-import com.comet.opik.domain.IdGeneratorImpl;
 import com.comet.opik.infrastructure.OpikConfiguration;
 import com.google.inject.Provides;
 import jakarta.inject.Singleton;
diff --git a/apps/opik-backend/src/main/java/com/comet/opik/infrastructure/log/UserFacingLoggingFactory.java b/apps/opik-backend/src/main/java/com/comet/opik/infrastructure/log/UserFacingLoggingFactory.java
@@ -3,6 +3,7 @@
 import ch.qos.logback.classic.AsyncAppender;
 import ch.qos.logback.classic.Logger;
 import ch.qos.logback.classic.LoggerContext;
+import com.comet.opik.infrastructure.OpikConfiguration;
 import com.comet.opik.infrastructure.log.tables.UserLogTableFactory;
 import io.r2dbc.spi.ConnectionFactory;
 import lombok.NonNull;
@@ -18,9 +19,9 @@ public class UserFacingLoggingFactory {
     private static AsyncAppender asyncAppender;
 
     public static synchronized void init(@NonNull ConnectionFactory connectionFactory, int batchSize,
-            @NonNull Duration flushIntervalSeconds) {
+            @NonNull Duration flushIntervalSeconds, @NonNull OpikConfiguration configuration) {
 
-        UserLogTableFactory tableFactory = UserLogTableFactory.getInstance(connectionFactory);
+        UserLogTableFactory tableFactory = UserLogTableFactory.getInstance(connectionFactory, configuration);
         ClickHouseAppender clickHouseAppender = ClickHouseAppender.init(tableFactory, batchSize, flushIntervalSeconds,
                 CONTEXT);
 
diff --git a/apps/opik-backend/src/main/java/com/comet/opik/infrastructure/log/tables/UserLogTableFactory.java b/apps/opik-backend/src/main/java/com/comet/opik/infrastructure/log/tables/UserLogTableFactory.java
@@ -3,6 +3,7 @@
 import ch.qos.logback.classic.spi.ILoggingEvent;
 import com.comet.opik.domain.evaluators.AutomationRuleEvaluatorLogsDAO;
 import com.comet.opik.domain.evaluators.UserLog;
+import com.comet.opik.infrastructure.OpikConfiguration;
 import io.r2dbc.spi.ConnectionFactory;
 import lombok.NonNull;
 import reactor.core.publisher.Mono;
@@ -12,8 +13,9 @@
 
 public interface UserLogTableFactory {
 
-    static UserLogTableFactory getInstance(ConnectionFactory factory) {
-        return new UserLogTableFactoryImpl(factory);
+    static UserLogTableFactory getInstance(@NonNull ConnectionFactory factory,
+            @NonNull OpikConfiguration configuration) {
+        return new UserLogTableFactoryImpl(factory, configuration);
     }
 
     interface UserLogTableDAO {
@@ -28,9 +30,9 @@ class UserLogTableFactoryImpl implements UserLogTableFactory {
 
     private final Map<UserLog, UserLogTableDAO> daoMap;
 
-    UserLogTableFactoryImpl(@NonNull ConnectionFactory factory) {
+    UserLogTableFactoryImpl(@NonNull ConnectionFactory factory, @NonNull OpikConfiguration configuration) {
         daoMap = Map.of(
-                UserLog.AUTOMATION_RULE_EVALUATOR, AutomationRuleEvaluatorLogsDAO.create(factory));
+                UserLog.AUTOMATION_RULE_EVALUATOR, AutomationRuleEvaluatorLogsDAO.create(factory, configuration));
     }
 
     @Override
diff --git a/apps/opik-backend/src/test/java/com/comet/opik/api/resources/v1/priv/DatasetsResourceIntegrationTest.java b/apps/opik-backend/src/test/java/com/comet/opik/api/resources/v1/priv/DatasetsResourceIntegrationTest.java
@@ -7,10 +7,10 @@
 import com.comet.opik.api.sorting.SortingFactoryDatasets;
 import com.comet.opik.domain.DatasetItemService;
 import com.comet.opik.domain.DatasetService;
-import com.comet.opik.domain.IdGeneratorImpl;
 import com.comet.opik.domain.Streamer;
 import com.comet.opik.domain.filter.FilterQueryBuilder;
 import com.comet.opik.infrastructure.auth.RequestContext;
+import com.comet.opik.infrastructure.db.IdGeneratorImpl;
 import com.comet.opik.infrastructure.json.JsonNodeMessageBodyWriter;
 import com.comet.opik.podam.PodamFactoryUtils;
 import com.comet.opik.utils.JsonUtils;
diff --git a/apps/opik-backend/src/test/java/com/comet/opik/domain/TraceServiceImplTest.java b/apps/opik-backend/src/test/java/com/comet/opik/domain/TraceServiceImplTest.java
@@ -5,6 +5,7 @@
 import com.comet.opik.api.sorting.TraceSortingFactory;
 import com.comet.opik.api.sorting.TraceThreadSortingFactory;
 import com.comet.opik.infrastructure.auth.RequestContext;
+import com.comet.opik.infrastructure.db.IdGeneratorImpl;
 import com.comet.opik.infrastructure.db.TransactionTemplateAsync;
 import com.comet.opik.infrastructure.lock.LockService;
 import com.comet.opik.utils.ErrorUtils;

Original file line number	Diff line number	Diff line change
`@@ -28,7 +28,7 @@ protected void configure() {`
`28`	`28`
`29`	`29`	`// Initialize the UserFacingRuleLollingFactory`
`30`	`30`	`UserFacingLoggingFactory.init(connectionFactory, clickHouseLogAppenderConfig.getBatchSize(),`
`31`		`- clickHouseLogAppenderConfig.getFlushIntervalDuration());`
	`31`	`+ clickHouseLogAppenderConfig.getFlushIntervalDuration(), configuration(OpikConfiguration.class));`
`32`	`32`	`}`
`33`	`33`
`34`	`34`	`@Provides`