Sort entities before batch update (for SQL)

application/src/main/resources/thingsboard.yml

@@ -284,6 +284,8 @@ sql:
       batch_max_delay: "${SQL_TS_LATEST_BATCH_MAX_DELAY_MS:100}"
       stats_print_interval_ms: "${SQL_TS_LATEST_BATCH_STATS_PRINT_MS:10000}"
       batch_threads: "${SQL_TS_LATEST_BATCH_THREADS:4}"
+    # Specify whether to sort entities before batch update. Should be enabled for cluster mode to avoid deadlocks
+    batch_sort: "${SQL_BATCH_SORT:false}"
     # Specify whether to remove null characters from strValue of attributes and timeseries before insert
     remove_null_chars: "${SQL_REMOVE_NULL_CHARS:true}"
   # Specify whether to log database queries and their parameters generated by entity query repository

dao/src/main/java/org/thingsboard/server/dao/sql/TbSqlBlockingQueue.java

@@ -22,6 +22,7 @@ import org.thingsboard.common.util.ThingsBoardThreadFactory;
 import org.thingsboard.server.common.stats.MessagesStats;
 
 import java.util.ArrayList;
+import java.util.Comparator;
 import java.util.List;
 import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.ExecutorService;
@@ -30,6 +31,7 @@ import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.TimeUnit;
 import java.util.function.Consumer;
 import java.util.stream.Collectors;
+import java.util.stream.Stream;
 
 @Slf4j
 public class TbSqlBlockingQueue<E> implements TbSqlQueue<E> {
@@ -46,7 +48,7 @@ public class TbSqlBlockingQueue<E> implements TbSqlQueue<E> {
     }
 
     @Override
-    public void init(ScheduledLogExecutorComponent logExecutor, Consumer<List<E>> saveFunction, int index) {
+    public void init(ScheduledLogExecutorComponent logExecutor, Consumer<List<E>> saveFunction, Comparator<E> batchUpdateComparator, int index) {
         executor = Executors.newSingleThreadExecutor(ThingsBoardThreadFactory.forName("sql-queue-" + index + "-" + params.getLogName().toLowerCase()));
         executor.submit(() -> {
             String logName = params.getLogName();
@@ -65,7 +67,11 @@ public class TbSqlBlockingQueue<E> implements TbSqlQueue<E> {
                     queue.drainTo(entities, batchSize - 1);
                     boolean fullPack = entities.size() == batchSize;
                     log.debug("[{}] Going to save {} entities", logName, entities.size());
-                    saveFunction.accept(entities.stream().map(TbSqlQueueElement::getEntity).collect(Collectors.toList()));
+                    Stream<E> entitiesStream = entities.stream().map(TbSqlQueueElement::getEntity);
+                    saveFunction.accept(
+                            (params.isBatchSortEnabled() ? entitiesStream.sorted(batchUpdateComparator) : entitiesStream)
+                                    .collect(Collectors.toList())
+                    );
                     entities.forEach(v -> v.getFuture().set(null));
                     stats.incrementSuccessful(entities.size());
                     if (!fullPack) {

dao/src/main/java/org/thingsboard/server/dao/sql/TbSqlBlockingQueueParams.java

@@ -31,4 +31,5 @@ public class TbSqlBlockingQueueParams {
     private final long maxDelay;
     private final long statsPrintIntervalMs;
     private final String statsNamePrefix;
+    private final boolean batchSortEnabled;
 }

dao/src/main/java/org/thingsboard/server/dao/sql/TbSqlBlockingQueueWrapper.java

@@ -21,6 +21,7 @@ import lombok.extern.slf4j.Slf4j;
 import org.thingsboard.server.common.stats.MessagesStats;
 import org.thingsboard.server.common.stats.StatsFactory;
 
+import java.util.Comparator;
 import java.util.List;
 import java.util.concurrent.CopyOnWriteArrayList;
 import java.util.function.Consumer;
@@ -36,12 +37,20 @@ public class TbSqlBlockingQueueWrapper<E> {
     private final int maxThreads;
     private final StatsFactory statsFactory;
 
-    public void init(ScheduledLogExecutorComponent logExecutor, Consumer<List<E>> saveFunction) {
+    /**
+     * Starts TbSqlBlockingQueues.
+     *
+     * @param  logExecutor  executor that will be printing logs and statistics
+     * @param  saveFunction function to save entities in database
+     * @param  batchUpdateComparator comparator to sort entities by primary key to avoid deadlocks in cluster mode
+     *                               NOTE: you must use all of primary key parts in your comparator
+     */
+    public void init(ScheduledLogExecutorComponent logExecutor, Consumer<List<E>> saveFunction, Comparator<E> batchUpdateComparator) {
         for (int i = 0; i < maxThreads; i++) {
             MessagesStats stats = statsFactory.createMessagesStats(params.getStatsNamePrefix() + ".queue." + i);
             TbSqlBlockingQueue<E> queue = new TbSqlBlockingQueue<>(params, stats);
             queues.add(queue);
-            queue.init(logExecutor, saveFunction, i);
+            queue.init(logExecutor, saveFunction, batchUpdateComparator, i);
         }
     }
 

dao/src/main/java/org/thingsboard/server/dao/sql/TbSqlQueue.java

@@ -17,12 +17,13 @@ package org.thingsboard.server.dao.sql;
 
 import com.google.common.util.concurrent.ListenableFuture;
 
+import java.util.Comparator;
 import java.util.List;
 import java.util.function.Consumer;
 
 public interface TbSqlQueue<E> {
 
-    void init(ScheduledLogExecutorComponent logExecutor, Consumer<List<E>> saveFunction, int queueIndex);
+    void init(ScheduledLogExecutorComponent logExecutor, Consumer<List<E>> saveFunction, Comparator<E> batchUpdateComparator, int queueIndex);
 
     void destroy();
 

dao/src/main/java/org/thingsboard/server/dao/sql/attributes/JpaAttributeDao.java

@@ -38,6 +38,7 @@ import org.thingsboard.server.dao.sql.TbSqlBlockingQueueWrapper;
 import javax.annotation.PostConstruct;
 import javax.annotation.PreDestroy;
 import java.util.Collection;
+import java.util.Comparator;
 import java.util.List;
 import java.util.Optional;
 import java.util.function.Function;
@@ -71,6 +72,9 @@ public class JpaAttributeDao extends JpaAbstractDaoListeningExecutorService impl
     @Value("${sql.attributes.batch_threads:4}")
     private int batchThreads;
 
+    @Value("${sql.batch_sort:false}")
+    private boolean batchSortEnabled;
+
     private TbSqlBlockingQueueWrapper<AttributeKvEntity> queue;
 
     @PostConstruct
@@ -81,11 +85,17 @@ public class JpaAttributeDao extends JpaAbstractDaoListeningExecutorService impl
                 .maxDelay(maxDelay)
                 .statsPrintIntervalMs(statsPrintIntervalMs)
                 .statsNamePrefix("attributes")
+                .batchSortEnabled(batchSortEnabled)
                 .build();
 
         Function<AttributeKvEntity, Integer> hashcodeFunction = entity -> entity.getId().getEntityId().hashCode();
         queue = new TbSqlBlockingQueueWrapper<>(params, hashcodeFunction, batchThreads, statsFactory);
-        queue.init(logExecutor, v -> attributeKvInsertRepository.saveOrUpdate(v));
+        queue.init(logExecutor, v -> attributeKvInsertRepository.saveOrUpdate(v),
+                Comparator.comparing((AttributeKvEntity attributeKvEntity) -> attributeKvEntity.getId().getEntityId())
+                        .thenComparing(attributeKvEntity -> attributeKvEntity.getId().getEntityType().name())
+                        .thenComparing(attributeKvEntity -> attributeKvEntity.getId().getAttributeType())
+                        .thenComparing(attributeKvEntity -> attributeKvEntity.getId().getAttributeKey())
+        );
     }
 
     @PreDestroy

dao/src/main/java/org/thingsboard/server/dao/sqlts/AbstractChunkedAggregationTimeseriesDao.java

@@ -21,6 +21,7 @@ import com.google.common.util.concurrent.MoreExecutors;
 import com.google.common.util.concurrent.SettableFuture;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Value;
 import org.springframework.data.domain.PageRequest;
 import org.springframework.data.domain.Sort;
 import org.thingsboard.server.common.data.id.EntityId;
@@ -31,6 +32,7 @@ import org.thingsboard.server.common.data.kv.ReadTsKvQuery;
 import org.thingsboard.server.common.data.kv.TsKvEntry;
 import org.thingsboard.server.common.stats.StatsFactory;
 import org.thingsboard.server.dao.DaoUtil;
+import org.thingsboard.server.dao.model.sql.AbstractTsKvEntity;
 import org.thingsboard.server.dao.model.sqlts.ts.TsKvEntity;
 import org.thingsboard.server.dao.sql.TbSqlBlockingQueueParams;
 import org.thingsboard.server.dao.sql.TbSqlBlockingQueueWrapper;
@@ -40,9 +42,7 @@ import org.thingsboard.server.dao.timeseries.TimeseriesDao;
 
 import javax.annotation.PostConstruct;
 import javax.annotation.PreDestroy;
-import java.util.ArrayList;
+import java.util.*;
-import java.util.List;
-import java.util.Optional;
 import java.util.concurrent.CompletableFuture;
 import java.util.function.Function;
 import java.util.stream.Collectors;
@@ -68,11 +68,16 @@ public abstract class AbstractChunkedAggregationTimeseriesDao extends AbstractSq
                 .maxDelay(tsMaxDelay)
                 .statsPrintIntervalMs(tsStatsPrintIntervalMs)
                 .statsNamePrefix("ts")
+                .batchSortEnabled(batchSortEnabled)
                 .build();
 
         Function<TsKvEntity, Integer> hashcodeFunction = entity -> entity.getEntityId().hashCode();
         tsQueue = new TbSqlBlockingQueueWrapper<>(tsParams, hashcodeFunction, tsBatchThreads, statsFactory);
-        tsQueue.init(logExecutor, v -> insertRepository.saveOrUpdate(v));
+        tsQueue.init(logExecutor, v -> insertRepository.saveOrUpdate(v),
+                Comparator.comparing((Function<TsKvEntity, UUID>) AbstractTsKvEntity::getEntityId)
+                        .thenComparing(AbstractTsKvEntity::getKey)
+                        .thenComparing(AbstractTsKvEntity::getTs)
+                );
     }
 
     @PreDestroy

dao/src/main/java/org/thingsboard/server/dao/sqlts/AbstractSqlTimeseriesDao.java

@@ -53,6 +53,9 @@ public abstract class AbstractSqlTimeseriesDao extends BaseAbstractSqlTimeseries
     @Value("${sql.timescale.batch_threads:4}")
     protected int timescaleBatchThreads;
 
+    @Value("${sql.batch_sort:false}")
+    protected boolean batchSortEnabled;
+
     protected ListenableFuture<List<TsKvEntry>> processFindAllAsync(TenantId tenantId, EntityId entityId, List<ReadTsKvQuery> queries) {
         List<ListenableFuture<List<TsKvEntry>>> futures = queries
                 .stream()

dao/src/main/java/org/thingsboard/server/dao/sqlts/SqlTimeseriesLatestDao.java

@@ -35,6 +35,7 @@ import org.thingsboard.server.common.data.kv.StringDataEntry;
 import org.thingsboard.server.common.data.kv.TsKvEntry;
 import org.thingsboard.server.common.stats.StatsFactory;
 import org.thingsboard.server.dao.DaoUtil;
+import org.thingsboard.server.dao.model.sql.AbstractTsKvEntity;
 import org.thingsboard.server.dao.model.sqlts.latest.TsKvLatestCompositeKey;
 import org.thingsboard.server.dao.model.sqlts.latest.TsKvLatestEntity;
 import org.thingsboard.server.dao.sql.ScheduledLogExecutorComponent;
@@ -50,12 +51,10 @@ import org.thingsboard.server.dao.util.SqlTsLatestAnyDao;
 import javax.annotation.Nullable;
 import javax.annotation.PostConstruct;
 import javax.annotation.PreDestroy;
-import java.util.ArrayList;
+import java.util.*;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-import java.util.Optional;
 import java.util.concurrent.ExecutionException;
+import java.util.function.Function;
+import java.util.stream.Collectors;
 
 @Slf4j
 @Component
@@ -90,6 +89,9 @@ public class SqlTimeseriesLatestDao extends BaseAbstractSqlTimeseriesDao impleme
     @Value("${sql.ts_latest.batch_threads:4}")
     private int tsLatestBatchThreads;
 
+    @Value("${sql.batch_sort:false}")
+    protected boolean batchSortEnabled;
+
     @Autowired
     protected ScheduledLogExecutorComponent logExecutor;
 
@@ -104,6 +106,7 @@ public class SqlTimeseriesLatestDao extends BaseAbstractSqlTimeseriesDao impleme
                 .maxDelay(tsLatestMaxDelay)
                 .statsPrintIntervalMs(tsLatestStatsPrintIntervalMs)
                 .statsNamePrefix("ts.latest")
+                .batchSortEnabled(false)
                 .build();
 
         java.util.function.Function<TsKvLatestEntity, Integer> hashcodeFunction = entity -> entity.getEntityId().hashCode();
@@ -118,9 +121,13 @@ public class SqlTimeseriesLatestDao extends BaseAbstractSqlTimeseriesDao impleme
                     trueLatest.put(key, ts);
                 }
             });
-            List<TsKvLatestEntity> latestEntities = new ArrayList<>(trueLatest.values());
+            List<TsKvLatestEntity> latestEntities = batchSortEnabled ?
+                    trueLatest.values().stream().sorted(Comparator.comparing((Function<TsKvLatestEntity, UUID>) AbstractTsKvEntity::getEntityId)
+                            .thenComparingInt(AbstractTsKvEntity::getKey)
+                    )
+                            .collect(Collectors.toList()) : new ArrayList<>(trueLatest.values());
             insertLatestTsRepository.saveOrUpdate(latestEntities);
-        });
+        }, (l, r) -> 0);
     }
 
     @PreDestroy

dao/src/main/java/org/thingsboard/server/dao/sqlts/timescale/TimescaleTimeseriesDao.java

@@ -33,6 +33,7 @@ import org.thingsboard.server.common.data.kv.ReadTsKvQuery;
 import org.thingsboard.server.common.data.kv.TsKvEntry;
 import org.thingsboard.server.common.stats.StatsFactory;
 import org.thingsboard.server.dao.DaoUtil;
+import org.thingsboard.server.dao.model.sql.AbstractTsKvEntity;
 import org.thingsboard.server.dao.model.sqlts.timescale.ts.TimescaleTsKvEntity;
 import org.thingsboard.server.dao.sql.TbSqlBlockingQueueParams;
 import org.thingsboard.server.dao.sql.TbSqlBlockingQueueWrapper;
@@ -43,11 +44,7 @@ import org.thingsboard.server.dao.util.TimescaleDBTsDao;
 
 import javax.annotation.PostConstruct;
 import javax.annotation.PreDestroy;
-import java.util.ArrayList;
+import java.util.*;
-import java.util.Collections;
-import java.util.List;
-import java.util.Optional;
-import java.util.UUID;
 import java.util.concurrent.CompletableFuture;
 import java.util.function.Function;
 
@@ -78,12 +75,17 @@ public class TimescaleTimeseriesDao extends AbstractSqlTimeseriesDao implements
                 .maxDelay(tsMaxDelay)
                 .statsPrintIntervalMs(tsStatsPrintIntervalMs)
                 .statsNamePrefix("ts.timescale")
+                .batchSortEnabled(batchSortEnabled)
                 .build();
 
         Function<TimescaleTsKvEntity, Integer> hashcodeFunction = entity -> entity.getEntityId().hashCode();
         tsQueue = new TbSqlBlockingQueueWrapper<>(tsParams, hashcodeFunction, timescaleBatchThreads, statsFactory);
 
-        tsQueue.init(logExecutor, v -> insertRepository.saveOrUpdate(v));
+        tsQueue.init(logExecutor, v -> insertRepository.saveOrUpdate(v),
+                Comparator.comparing((Function<TimescaleTsKvEntity, UUID>) AbstractTsKvEntity::getEntityId)
+                        .thenComparing(AbstractTsKvEntity::getKey)
+                        .thenComparing(AbstractTsKvEntity::getTs)
+        );
     }
 
     @PreDestroy