Aggregation function fixes to the SQL DAO layer

dao/src/main/java/org/thingsboard/server/dao/model/sql/TsKvEntity.java

@@ -34,7 +34,9 @@ public final class TsKvEntity implements ToData<TsKvEntry> {
     }
 
     public TsKvEntity(Double avgLongValue, Double avgDoubleValue) {
-        this.longValue = avgLongValue.longValue();
+        if(avgLongValue != null) {
+            this.longValue = avgLongValue.longValue();
+        }
         this.doubleValue = avgDoubleValue;
     }
 
@@ -104,4 +106,8 @@ public final class TsKvEntity implements ToData<TsKvEntry> {
         }
         return new BasicTsKvEntry(ts, kvEntry);
     }
+
+    public boolean isNotEmpty() {
+        return strValue != null || longValue != null || doubleValue != null || booleanValue != null;
+    }
 }

dao/src/main/java/org/thingsboard/server/dao/sql/timeseries/TsKvRepository.java

@@ -19,12 +19,14 @@ import org.springframework.data.domain.Pageable;
 import org.springframework.data.jpa.repository.Query;
 import org.springframework.data.repository.CrudRepository;
 import org.springframework.data.repository.query.Param;
+import org.springframework.scheduling.annotation.Async;
 import org.thingsboard.server.common.data.EntityType;
 import org.thingsboard.server.dao.model.sql.TsKvCompositeKey;
 import org.thingsboard.server.dao.model.sql.TsKvEntity;
 import org.thingsboard.server.dao.util.SqlDao;
 
 import java.util.List;
+import java.util.concurrent.Future;
 
 @SqlDao
 public interface TsKvRepository extends CrudRepository<TsKvEntity, TsKvCompositeKey> {
@@ -39,49 +41,52 @@ public interface TsKvRepository extends CrudRepository<TsKvEntity, TsKvComposite
                                       @Param("endTs") long endTs,
                                       Pageable pageable);
 
+    @Async
     @Query("SELECT new TsKvEntity(MAX(tskv.strValue), MAX(tskv.longValue), MAX(tskv.doubleValue)) FROM TsKvEntity tskv " +
             "WHERE tskv.entityId = :entityId AND tskv.entityType = :entityType " +
             "AND tskv.key = :entityKey AND tskv.ts > :startTs AND tskv.ts < :endTs")
-    TsKvEntity findMax(@Param("entityId") String entityId,
+    Future<TsKvEntity> findMax(@Param("entityId") String entityId,
                        @Param("entityType") EntityType entityType,
                        @Param("entityKey") String entityKey,
                        @Param("startTs") long startTs,
                        @Param("endTs") long endTs);
 
+    @Async
     @Query("SELECT new TsKvEntity(MIN(tskv.strValue), MIN(tskv.longValue), MIN(tskv.doubleValue)) FROM TsKvEntity tskv " +
             "WHERE tskv.entityId = :entityId AND tskv.entityType = :entityType " +
             "AND tskv.key = :entityKey AND tskv.ts > :startTs AND tskv.ts < :endTs")
-    TsKvEntity findMin(@Param("entityId") String entityId,
+    Future<TsKvEntity> findMin(@Param("entityId") String entityId,
                        @Param("entityType") EntityType entityType,
                        @Param("entityKey") String entityKey,
                        @Param("startTs") long startTs,
                        @Param("endTs") long endTs);
 
-
+    @Async
     @Query("SELECT new TsKvEntity(COUNT(tskv.booleanValue), COUNT(tskv.strValue), COUNT(tskv.longValue), COUNT(tskv.doubleValue)) FROM TsKvEntity tskv " +
             "WHERE tskv.entityId = :entityId AND tskv.entityType = :entityType " +
             "AND tskv.key = :entityKey AND tskv.ts > :startTs AND tskv.ts < :endTs")
-    TsKvEntity findCount(@Param("entityId") String entityId,
+    Future<TsKvEntity> findCount(@Param("entityId") String entityId,
-                         @Param("entityType") EntityType entityType,
+                                 @Param("entityType") EntityType entityType,
-                         @Param("entityKey") String entityKey,
+                                 @Param("entityKey") String entityKey,
-                         @Param("startTs") long startTs,
+                                 @Param("startTs") long startTs,
-                         @Param("endTs") long endTs);
+                                 @Param("endTs") long endTs);
-
 
+    @Async
     @Query("SELECT new TsKvEntity(AVG(tskv.longValue), AVG(tskv.doubleValue)) FROM TsKvEntity tskv " +
             "WHERE tskv.entityId = :entityId AND tskv.entityType = :entityType " +
             "AND tskv.key = :entityKey AND tskv.ts > :startTs AND tskv.ts < :endTs")
-    TsKvEntity findAvg(@Param("entityId") String entityId,
+    Future<TsKvEntity> findAvg(@Param("entityId") String entityId,
                        @Param("entityType") EntityType entityType,
                        @Param("entityKey") String entityKey,
                        @Param("startTs") long startTs,
                        @Param("endTs") long endTs);
 
 
+    @Async
     @Query("SELECT new TsKvEntity(SUM(tskv.longValue), SUM(tskv.doubleValue)) FROM TsKvEntity tskv " +
             "WHERE tskv.entityId = :entityId AND tskv.entityType = :entityType " +
             "AND tskv.key = :entityKey AND tskv.ts > :startTs AND tskv.ts < :endTs")
-    TsKvEntity findSum(@Param("entityId") String entityId,
+    Future<TsKvEntity> findSum(@Param("entityId") String entityId,
                        @Param("entityType") EntityType entityType,
                        @Param("entityKey") String entityKey,
                        @Param("startTs") long startTs,

extensions-core/src/main/java/org/thingsboard/server/extensions/core/plugin/telemetry/sub/SubscriptionUpdate.java

@@ -32,11 +32,7 @@ public class SubscriptionUpdate {
         this.subscriptionId = subscriptionId;
         this.data = new TreeMap<>();
         for (TsKvEntry tsEntry : data) {
-            List<Object> values = this.data.get(tsEntry.getKey());
+            List<Object> values = this.data.computeIfAbsent(tsEntry.getKey(), k -> new ArrayList<>());
-            if (values == null) {
-                values = new ArrayList<>();
-                this.data.put(tsEntry.getKey(), values);
-            }
             Object[] value = new Object[2];
             value[0] = tsEntry.getTs();
             value[1] = tsEntry.getValueAsString();